fix!: le registre d-ouverture n-a plus qu-un seul sens, et il cesse de mentir

Trois constats, un seul defaut sous trois visages : un etat enregistre comme vrai
qui ne l'est pas, sans moyen de se corriger. Deux des trois vivaient dans
`open-repo.ts`. Cinq tours sont alles dans l'appelant ; c'est le teneur de
registre qui mentait.

CE QUE LE REGISTRE VEUT DIRE, ecrit dans l'en-tete. `held` est le SEUL registre
d'ouverture : cette session detient un abonnement vivant a ce document. Il y avait
a cote un `opened` voulant dire « on a appele `doc_subscribe` une fois et on a
cesse d'attendre », et ce sens plus lache mentait de trois facons : il survivait a
un abonnement REFUSE par le courtier (donc la reprise que `subscribe.ts` offre
etait bloquee a jamais), il survivait a une reinitialisation (donc une nouvelle
session court-circuitait sur l'enregistrement de la precedente), et il laissait un
document annoncer une barriere jamais franchie. `opened` est supprime.

Les trois consequences, desormais vraies par construction :
- un abonnement qui a ECHOUE n'enregistre rien - `onSetupFailed`, le canal que
  `subscribe.ts` offre exactement pour cela, est enfin branche - donc l'appelant
  suivant s'abonne a nouveau ;
- une reinitialisation efface l'enregistrement ET DECLASSE l'etat de synchro :
  une barriere franchie dans la session qui vient de disparaitre n'en est pas une
  que celle-ci a franchie ;
- `"unknown"` ne veut plus dire qu'une chose : aucune barriere n'etait attendue ici
  (jamais demande, ou un `ng` sans `doc_subscribe`). Jamais « on a oublie ».

Cote appelant, un seul changement, et c'est le troisieme constat : quand un
document nomme n'a pas repondu, `publish` PUBLIE « la question reste ouverte » au
lieu de sortir en laissant en place l'instantane precedent. Sortir laissait un
refus DEJA LEVE a l'affichage : l'application lisait « vous ne detenez pas la
capacite de ce document » a propos d'un document qu'elle detient desormais.

Deux tests portaient l'ancien sens et sont corriges avec leur raison : celui du
`ng` sans `doc_subscribe` (le compteur d'ids du faux courtier redemarrait a chaque
test, donc deux tests aliasaient leurs depots - la lecon de `wallet-fake` a
l'echelle du fichier), et celui de la lecture froide, ou « timed-out » est
desormais ce que dit « cette page ne l'a pas confirme ».
This commit is contained in:
Sylvain Duchesne
2026-09-24 11:34:10 +02:00
parent 2571f8e738
commit 6b2b8ab42a
6 changed files with 218 additions and 32 deletions
@@ -76,7 +76,7 @@ import { mustNotAttempt } from "./reach";
import { fetchReadCap } from "./public-store";
import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap";
import { resubscribeDocs, subscribeDocUnguarded, type Unsubscribe } from "../surface/subscribe";
import { logStage, shortNuri } from "../shared-wallet/access-log";
import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log";
import type { Nuri } from "../model/types";
/**
@@ -89,8 +89,26 @@ import type { Nuri } from "../model/types";
// @provenance SyncState kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — the first `State` push IS the barrier; `timed-out` names the bounded fallback and is deliberately not `synced`
export type SyncState = "syncing" | "synced" | "timed-out";
/** Repos whose bootstrap open has completed (first `State` received OR timed out). */
const opened = new Set<Nuri>();
/*
* WHAT THIS REGISTRY MEANS, and it is the thing that went wrong five rounds running.
*
* `held` is the ONE record of openness: **this session holds a live bootstrap subscription
* to this document**. Nothing else records it, because two records of the same fact can
* disagree and this pair did. There was an `opened` set beside it meaning "we once called
* `doc_subscribe` and stopped waiting", and that looser meaning is what lied three ways:
* it survived a subscribe the broker REJECTED (so the retry `subscribe.ts` offers was
* blocked for ever), it survived a registry reset (so a new session short-circuited on the
* previous one's record), and it let a document report a barrier it had never crossed.
*
* The three consequences, each now true by construction:
* - a subscribe that FAILED records nothing, so the next caller subscribes again;
* - a reset clears the record AND downgrades the sync state, so nothing claims authority
* the new session has not earned;
* - `syncState` says `"unknown"` for one reason only: no barrier was ever expected here
* (never requested, or an `ng` with no `doc_subscribe` and so no `State` semantics).
* It never means "we forgot".
*/
/** In-flight opens... see {@link inFlight} below; `held` is the record of openness. */
/** In-flight opens, so concurrent `ensureRepoOpen(nuri)` share one subscription. */
const inFlight = new Map<Nuri, Promise<void>>();
/**
@@ -162,18 +180,22 @@ export function resetOpenedRepos(): void {
// nothing an absence. The abandoned open's own fallback fires later and sets the same
// value, so the two cannot disagree.
const abandoned = [...settleInFlight.entries()];
// Every document this reset takes a subscription away from, in flight or long settled.
// Their sync state is DOWNGRADED rather than cleared: a barrier crossed in the session
// that just went away is not one this session has crossed, and clearing would leave them
// reading `"unknown"` - which means "no barrier was ever expected here" and is therefore
// treated as authoritative. That conflation is how a read that had been correctly
// withheld came back claiming a success, against a query nothing had re-subscribed.
const disowned = new Set<Nuri>([...held.keys(), ...abandoned.map(([nuri]) => nuri)]);
// Everything in flight now belongs to a registry that no longer exists, so its
// continuation must write nothing - see {@link registryEpoch}.
registryEpoch += 1;
opened.clear();
inFlight.clear();
settleInFlight.clear();
held.clear();
syncState.clear();
for (const [nuri, settle] of abandoned) {
syncState.set(nuri, "timed-out");
settle();
}
for (const nuri of disowned) syncState.set(nuri, "timed-out");
for (const [, settle] of abandoned) settle();
boundSessionId = null;
OPEN_TIMEOUT_MS = 8000;
// Whatever ELSE was still subscribed to a document - an inbox observation, an
@@ -272,9 +294,12 @@ export async function ensureRepoOpen(nuri: Nuri, holderKey?: string): Promise<vo
export async function openRepoUnguarded(nuri: Nuri): Promise<void> {
// Drop the registry if the session changed (in-page re-login → fresh verifier).
await syncSession();
if (opened.has(nuri)) return;
// IN FLIGHT first, so a concurrent caller joins the open and waits on the same barrier
// rather than stepping past it: `held` is set the moment the subscription is taken out,
// which is before the `State` that makes it authoritative.
const pending = inFlight.get(nuri);
if (pending) return pending;
if (held.has(nuri)) 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`
@@ -282,7 +307,9 @@ export async function openRepoUnguarded(nuri: Nuri): Promise<void> {
// path has no barrier semantics, and claiming "synced" here would be a lie.
const ng = getConfig().ng as { doc_subscribe?: unknown };
if (typeof ng.doc_subscribe !== "function") {
opened.add(nuri);
// Nothing to open and nothing to record: there is no subscription to hold, so `held`
// must stay empty for this document or it would claim one. No sync state either -
// `"unknown"` is exactly right here, and it is the one place that MEANS it.
return;
}
@@ -338,17 +365,41 @@ export async function openRepoUnguarded(nuri: Nuri): Promise<void> {
// Unguarded on purpose: the caller already decided. `ensureRepoOpen` applied
// rule 2 above; `ensurePhysicalRepoOpen` is the machinery's door and is not
// subject to the boundary at all (see physical.ts).
const unsub = subscribeDocUnguarded(nuri, (_r, type) => {
if (type === "State") onState();
});
// A failed establish is REPORTED, and `subscribe.ts` offers this channel precisely
// so that a caller can retry. Ignoring it is what made a refused subscribe look like
// an open repo: the fallback below fired, the open was recorded, and every later
// attempt short-circuited on a subscription that had never existed.
const onSetupFailed = (error: unknown): void => {
if (myEpoch !== registryEpoch) return;
console.error(accessLogPrefix() + " ensureRepoOpen: subscribe failed for " + shortNuri(nuri) + ":", error);
// Record NOTHING, so the next caller subscribes again - and say plainly that no
// barrier was crossed, rather than leaving a state that reads as authority.
const taken = held.get(nuri);
if (taken) {
held.delete(nuri);
try {
taken();
} catch {
/* ignore */
}
}
syncState.set(nuri, "timed-out");
settleInFlight.get(nuri)?.();
};
const unsub = subscribeDocUnguarded(
nuri,
(_r, type) => {
if (type === "State") onState();
},
onSetupFailed,
);
held.set(nuri, unsub);
setTimeout(onTimeout, OPEN_TIMEOUT_MS);
});
// Only if this open still belongs to the registry it started in. After a reset the
// three maps below are somebody else's, and `opened` in particular would become a
// permanent lie: nothing re-subscribes a repo that is already recorded as open.
// Only if this open still belongs to the registry it started in. After a reset these
// maps are somebody else's, and writing into them is how a new session came to
// short-circuit on the previous one's record.
if (myEpoch !== registryEpoch) return;
opened.add(nuri);
inFlight.delete(nuri);
settleInFlight.delete(nuri);
})();
@@ -380,6 +380,26 @@ export function useShapeQuery<T extends BaseType = BaseType>(
});
}
/**
* Publish "the question is still open" - no authority yet, and nothing to say about the
* data either way.
*
* `data` is carried over for the same reason {@link setUnknown} carries it: an observable
* cannot un-emit a list a subscriber has already rendered. Before the first answer there
* is nothing to carry and it stays `[]`. Idempotent, so a document that keeps failing to
* sync does not wake every bound view on each attempt.
*/
function setNotYetAnswered(): void {
if (snapshot.isPending && !snapshot.isSuccess && !snapshot.isError) return;
setSnapshot({
data: snapshot.data,
isPending: true,
isSuccess: false,
isError: false,
error: undefined,
});
}
/**
* The readable items of the live subscription, as a plain array.
*
@@ -470,14 +490,20 @@ export function useShapeQuery<T extends BaseType = BaseType>(
setUnknown(refusal);
return;
}
// NOT AUTHORITATIVE YET, so say nothing rather than say "empty". A named document
// whose sync barrier was never crossed may hold anything, and publishing the ORM's
// view of it as `isSuccess` is the original defect in the one case the barrier was
// added to close. Returning here leaves the snapshot where it was - the initial
// pending one on a first publication - which is the contract's "a longer `isPending`",
// and never a successful empty. Re-read on every publication, so a `State` that lands
// later and brings data republishes through the ORM's own change signal and answers.
if (!namedDocumentsArrived()) return;
// NOT AUTHORITATIVE YET, so say the question is still open rather than say "empty". A
// named document whose sync barrier was never crossed may hold anything, and
// publishing the ORM's view of it as `isSuccess` is the original defect in the one case
// the barrier was added to close.
//
// It PUBLISHES pending rather than returning. Returning left whatever snapshot was
// standing, and that was wrong in the case that matters: a refusal that has since been
// LIFTED stayed on display, so an application was handed `isError` reading "you do not
// hold this document's cap" about a document it now holds. Whatever the state was, it
// is not the answer to the question just asked.
if (!namedDocumentsArrived()) {
setNotYetAnswered();
return;
}
setSnapshot({
data: readable(handle),
isPending: false,
@@ -113,8 +113,18 @@ describe("reading the messages left on my note, on a page that has just loaded",
// Not a failure anyone can see: the message is on the broker, Alice owns the inbox,
// and the call returns a perfectly ordinary empty list.
expect(await inboxSurface.readForDocument(note)).toEqual([]);
// …because nothing ever brought the NOTE into view. The address lives on it.
expect(getSyncState(note)).toBe("unknown");
// ...because nothing on THIS page ever brought the NOTE into view. The address lives
// on it.
//
// `"timed-out"`, not `"unknown"`, since 2026-09-24: a registry reset now DOWNGRADES the
// documents it takes subscriptions away from instead of forgetting their state, so
// `"unknown"` means only "no barrier was ever expected here" and can be read as
// authoritative. Note this particular value is a harness artefact and not what a real
// reload produces: `reloadPage` resets the registry inside one process, so the note is
// still in it from the first page, where a browser reload would start with an empty
// module and answer `"unknown"`. Either way it is the safe reading - neither value lets
// a zero-row read be called a definitive absence here.
expect(getSyncState(note)).toBe("timed-out");
});
test("the synced document-addressed read answers the message, over that same state", async () => {
+26 -2
View File
@@ -20,7 +20,7 @@
*/
import { describe, it, expect, mock, beforeEach, afterAll } from "bun:test";
import { ensureRepoOpen, getSyncState, resetOpenedRepos } from "../src/emulated-verifier/open-repo";
import { ensureRepoOpen, getSyncState, resetOpenedRepos, setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
import { readDocumentSubjects } from "../src/surface/read-document";
import { configure } from "../src/index";
import { configureStoreRegistry, setCurrentUser } from "../src/shared-wallet/bootstrap";
@@ -55,7 +55,7 @@ const FP = "http://festipod.org/";
function makeFakeNgWithSubscribe(
triplesByDoc: Record<string, Array<[string, string]>>,
options: { pushState?: boolean } = {},
options: { pushState?: boolean; subscribeFails?: boolean } = {},
) {
// A broker that accepts the subscription and never pushes is the state the bounded
// fallback exists for, and the only way to hold an open genuinely IN FLIGHT.
@@ -67,6 +67,9 @@ function makeFakeNgWithSubscribe(
const doc_subscribe = mock(async (nuri: string, _sid: string, cb: (r: unknown) => void) => {
subscribed.add(nuri);
subscribeCallOrder.push(nuri);
// A broker that REFUSES the subscription - a real failure, and one `subscribe.ts`
// deliberately makes retryable by offering `onSetupFailed`.
if (options.subscribeFails) throw new Error("BrokerError: cannot subscribe");
// Simulate initial State push (synchronously deferred so the subscription
// setup promise path in ensureRepoOpen can resolve it).
if (pushState) setTimeout(() => cb({ V0: { State: {} } }), 0);
@@ -237,6 +240,27 @@ describe("ensureRepoOpen", () => {
await reopened;
});
// A subscribe the broker REFUSED leaves nothing behind, so it can be tried again.
//
// `subscribe.ts` reports a failed establish through `onSetupFailed` precisely so a
// caller can retry; `open-repo` ignored it, waited out the bounded fallback, and then
// recorded the repo as OPEN. Every later open short-circuited on that record, so
// `refetch()` issued no new subscribe and the read had no exit at all.
it("a subscribe the broker REFUSED is not recorded as open, so it can be retried", async () => {
const ng = makeFakeNgWithSubscribe({}, { subscribeFails: true });
inject(ng);
setOpenTimeoutForTests(30);
await ensureRepoOpen("did:ng:o:refused-sub");
expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
// Not authoritative: no `State` was ever going to arrive.
expect(getSyncState("did:ng:o:refused-sub")).toBe("timed-out");
// The retry a caller is entitled to make.
await ensureRepoOpen("did:ng:o:refused-sub");
expect(ng.doc_subscribe).toHaveBeenCalledTimes(2);
});
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"]],
+75 -2
View File
@@ -26,6 +26,7 @@ import { useShapeQuery } from "../src/surface/use-shape-query";
import { configure } from "../src/index";
import { configureStoreRegistry, setCurrentUser, getCaps } from "../src/shared-wallet/bootstrap";
import { resetCaps, resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap";
import { resetOpenedRepos as resetRegistry } from "../src/emulated-verifier/open-repo";
import { resetRegistryCache, createEntityDoc } from "../src/shared-wallet/account-registry";
import { getSyncState, resetOpenedRepos, setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
import { mintCap } from "../src/emulated-verifier/caps";
@@ -59,6 +60,16 @@ interface SubRec { nuri: string; cb: (r: unknown) => void }
* The stateful fake `ng` - the shim account, the per-scope index and the document
* creation the page does around the read. It is NOT what answers the read any more.
*/
/**
* The repo ids this file's fake broker has handed out - MONOTONIC across every `makeFake`,
* for the reason `wallet-fake.ts` records at its own scale: a broker never mints a repo id
* twice, because an id is a public key. A counter INSIDE the builder restarted at each
* test, so two tests in this file minted the same nuri and aliased each other's entries in
* the process-global open registry - invisible while a reset cleared every document's sync
* state, and immediately visible once a reset began downgrading it instead.
*/
let mintedHere = 0;
function makeFake(options: { pushState?: boolean; subscribe?: boolean } = {}) {
/** Anchors whose anchored read throws. Mutable after boot, so a suite builds a healthy
* world first and breaks only the one call it is about. */
@@ -69,7 +80,6 @@ function makeFake(options: { pushState?: boolean; subscribe?: boolean } = {}) {
// without waiting for it.
const pushState = options.pushState !== false;
const quads: Quad[] = [];
let docCounter = 0;
const subs: SubRec[] = [];
// A NAMESPACE OF ITS OWN, and it is not cosmetic. `wallet-fake.ts` mints into
@@ -79,7 +89,7 @@ function makeFake(options: { pushState?: boolean; subscribe?: boolean } = {}) {
// answers, hanging it until the 8s fallback. `wallet-fake` carries the same lesson at its
// own scale ("a broker never mints a repo id twice - an id is a public key - so neither
// does this"); this is that lesson one scope wider.
const doc_create = mock(async () => `did:ng:o:usq${++docCounter}`);
const doc_create = mock(async () => `did:ng:o:usq${++mintedHere}`);
const sparql_update = mock(async (...a: unknown[]) => {
const query = a[1] as string;
@@ -1168,6 +1178,69 @@ describe("useShapeQuery", () => {
unsub();
});
// -- what the REGISTRY says, and whether it is true -----------------------
//
// Both of these are the same fault seen from the caller: state recorded as authoritative
// that is not, with no way to correct itself. The repair is in the bookkeeper, not here.
it("(ee) a session change does not turn a withheld read into a success", async () => {
const ng = makeFake({ pushState: false });
const orm = makeOrm();
inject(ng, orm);
setOpenTimeoutForTests(30);
setCurrentUser("vic");
const mine = await createEntityDoc("vic", "protected");
const obs = useShapeQuery(EventShape, { graphs: [mine] });
const unsub = obs.subscribe(() => {});
for (let i = 0; i < 40 && getSyncState(mine) !== "timed-out"; i++) await tick();
expect(obs.getSnapshot().isSuccess).toBe(false); // correctly withheld
// An in-page re-login: `syncSession` resets the registry when the session id changes,
// and `restoreOwnCaps` / `readUserStore` reach it at connect. The registry used to
// clear every document's sync state, and a cleared state reads `"unknown"` - which is
// deliberately treated as authoritative, because on an `ng` with no `doc_subscribe`
// there is no barrier to have missed. So a document that was KNOWN not to have synced
// came back claiming authority it never had, against a query this session never
// re-subscribed.
resetRegistry();
getCaps().learn(mintCap("did:ng:o:some-arrival"));
for (let i = 0; i < 20; i++) await tick();
expect(obs.getSnapshot().isSuccess).toBe(false);
unsub();
});
it("(ff) a refusal that has been lifted is never published again as the answer", async () => {
const ng = makeFake({ pushState: false });
const orm = makeOrm();
inject(ng, orm);
setOpenTimeoutForTests(30);
setCurrentUser("alice");
const hers = await createEntityDoc("alice", "protected");
setCurrentUser("walt");
const obs = useShapeQuery(EventShape, { graphs: [hers] });
const unsub = obs.subscribe(() => {});
for (let i = 0; i < 20 && !obs.getSnapshot().isError; i++) await tick();
expect(String((obs.getSnapshot().error as Error).message)).toContain("does not hold");
// The cap arrives, so the refusal is LIFTED - and the document then fails to sync.
// Two things must follow, and neither did: the answer is `isPending` (this round
// decided that a barrier not crossed is not a failure), and the application must not
// be handed a reason that is no longer true. It was still reading "does not hold this
// document's cap" about a document Walt now holds.
getCaps().learn(mintCap(hers));
for (let i = 0; i < 40 && obs.getSnapshot().isError; i++) await tick();
expect(getSyncState(hers)).toBe("timed-out");
const snap = obs.getSnapshot();
expect(snap.isError).toBe(false);
expect(snap.isSuccess).toBe(false);
expect(snap.isPending).toBe(true);
unsub();
});
// -- teardown ------------------------------------------------------------
it("(k) the last unsubscribe closes the shape query", async () => {