WIP: open/subscribe repos before anchored cold-start reads (branche, non mergée)

Défaut visé : sur une session verifier fraîche (reconnexion), la lecture ancrée
tape des repos pas encore ouverts dans self.repos → 0 ligne. Nouveau module
`open-repo.ts` : `ensureReposOpen`/`ensureRepoOpen` ouvrent/souscrivent un repo
via la primitive existante `subscribeDoc` (doc_subscribe) et attendent le push
d'état initial (borné, sans polling) AVANT la lecture ancrée. Câblé en amont de
`readUnion` (read-model) et `readScopeIndex` (store-registry). Idempotent par
session (Set des NURI ouverts + Map in-flight ; ré-ouverture si la session
injectée change). No-op si le `ng` injecté n'a pas doc_subscribe (fake unitaire).

État — NON MERGÉ, incomplet :
- AIDE le PROTECTED : la participation remonte au cold-start (mesuré app, timing
  un peu bruité).
- N'ADRESSE PAS l'accueil PUBLIC : `readScopeIndex` de l'index public rend 0 même
  côté écrivain même-session — le fix ouvre le repo mais l'index reste vide. Le
  code d'index lib est prouvé scope-symétrique, donc la cause du public est
  ailleurs (broker/store ou chemin app), À MESURER SOUS BROKER (actuellement
  injoignable). Nécessité du fix elle-même non prouvée en e2e lib (broker down).

gate broker-indépendant : tsc --noEmit propre ; bun test 91 pass (worker).
Inclut le scaffolding e2e de repro reconnexion (broker/run/sdk-entry).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sylvain Duchesne
2026-07-08 10:46:20 +02:00
parent d8c36bac3b
commit 1825d4d72f
6 changed files with 402 additions and 6 deletions
+158
View File
@@ -0,0 +1,158 @@
/**
* open-repo — cold-start repo opening for the ANCHORED read path (polyfill-era).
*
* ── The cold-start defect this heals ──────────────────────────────────────
* The anchored read path (`read-model.ts` `readDoc`, `store-registry.ts`
* `readScopeIndex`) assumes the target repo is already in the verifier's
* `self.repos` — true within the session that CREATED the doc (every `doc_create`
* opens it), but FALSE on a FRESH session over the same persistent wallet
* (reconnection / new page / re-login). On that fresh session nothing has opened
* the user's scope-index or entity repos yet, so an anchored `sparql_query`
* resolves a repo absent from `self.repos` and silently returns 0 rows (never a
* `RepoNotFound`) — persisted documents read as empty.
*
* The circularity that made this self-inflicted: `doc_subscribe` WOULD open the
* repo, but the reactive layer only subscribes AFTER the listing produced NURIs —
* and the listing (`readScopeIndex`) is itself an anchored read of a not-yet-open
* index repo → 0 rows → nothing to subscribe → nothing ever opens. Verified fix
* (adversarial pass): on a fresh session, `doc_subscribe(<docNuri>)` THEN the
* anchored re-read returns the data. So we OPEN the repo before the anchored read.
*
* ── How we open ───────────────────────────────────────────────────────────
* We reuse the existing per-document primitive {@link subscribeDoc} (the typed
* wrapper over the platform's `doc_subscribe`) — NOT a parallel channel. The
* subscribe pushes the repo's initial `State` once it is opened/loaded; we await
* that first push (bounded) and treat it as "the repo is now in the session". The
* subscription is kept ALIVE for the whole session (that is what keeps the repo
* open) — it is a bootstrap open, distinct from any reactive subscription a caller
* later establishes for change signals.
*
* ── Idempotence / perf (once per session, no polling) ─────────────────────
* The registry opens each repo at most ONCE per session: `opened` records completed
* opens (a hit skips everything), `inFlight` de-dupes concurrent opens of the same
* repo. A brand-new page / module instance starts with an empty registry; and when
* the injected session id CHANGES within the same page (an in-page re-login /
* `session_stop`+`session_start`, whose new verifier has an empty `self.repos`) the
* registry auto-resets (`syncSession`) so repos are re-opened against the new session
* rather than wrongly skipped as "already open". No polling: we wait on the
* initial-state push, with a bounded fallback timeout so a missing push can't hang.
*
* ── Migration ─────────────────────────────────────────────────────────────
* At the real multi-store migration this becomes "open the user's store repo by
* cap" (a native broker fetch) done once at bootstrap; the anchored read then
* resolves a same-session repo directly. Polyfill-era, removed with the shim.
*/
import { getConfig, getStoreRegistryDeps } from "./polyfill";
import { subscribeDoc, type Unsubscribe } from "./subscribe";
import type { Nuri } from "./types";
/** Repos whose bootstrap open has completed (initial state pushed or timed out). */
const opened = new Set<Nuri>();
/** In-flight opens, so concurrent `ensureRepoOpen(nuri)` share one subscription. */
const inFlight = new Map<Nuri, Promise<void>>();
/** Live bootstrap subscriptions, kept for the session (this is what holds repos open). */
const held = new Map<Nuri, Unsubscribe>();
/** The session id the current `opened`/`held` entries belong to. A change means a
* new verifier session (fresh `self.repos`) → the registry must be invalidated. */
let boundSessionId: string | number | null = null;
/**
* 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
* pathological case (a doc that never pushes), so a read is never blocked forever —
* it just proceeds (and yields 0 rows, exactly as before, for a genuinely-absent doc).
*/
const OPEN_TIMEOUT_MS = 8000;
/** Reset the open registry (mainly for tests / a switched wallet). Tears down the
* held bootstrap subscriptions so a subsequent open re-subscribes cleanly. */
export function resetOpenedRepos(): void {
for (const unsub of held.values()) {
try {
unsub();
} catch {
/* ignore */
}
}
opened.clear();
inFlight.clear();
held.clear();
boundSessionId = null;
}
/**
* Invalidate the registry if the active session id changed since it was populated.
* A new session id means a new verifier with an EMPTY `self.repos`, so entries from
* the previous session must NOT suppress re-opening under the new one. Tolerant: if
* the session can't be resolved, keep the current registry (best effort).
*/
async function syncSession(): Promise<void> {
let sid: string | number | null = null;
try {
sid = (await getStoreRegistryDeps().getSession()).sessionId;
} catch {
return; // no session deps wired (unit fake path) — nothing to invalidate against
}
if (boundSessionId !== null && boundSessionId !== sid) resetOpenedRepos();
boundSessionId = sid;
}
/**
* Ensure `nuri`'s repo is OPEN in the current session before an anchored read,
* so the cold-start (fresh session, same persistent wallet) resolves it instead
* of returning 0 rows. Opens via {@link subscribeDoc} and awaits the initial state
* push (bounded). Idempotent: a repo already opened (or in flight) is not re-opened.
*
* Tolerant by construction: if the injected `ng` exposes no `doc_subscribe` (e.g.
* the fake `ng` in the unit suite), this is a no-op — the read proceeds unchanged.
* Never throws; a failed open just leaves the read to behave as it did before.
*/
export async function ensureRepoOpen(nuri: Nuri): Promise<void> {
if (!nuri) return;
// Drop the registry if the session changed (in-page re-login → fresh verifier).
await syncSession();
if (opened.has(nuri)) return;
const pending = inFlight.get(nuri);
if (pending) return pending;
// No reactive primitive on the injected ng (fake-ng unit suite): nothing to open.
const ng = getConfig().ng as { doc_subscribe?: unknown };
if (typeof ng.doc_subscribe !== "function") {
opened.add(nuri);
return;
}
const p = (async () => {
await new Promise<void>((resolve) => {
let settled = false;
const done = (): void => {
if (settled) return;
settled = true;
resolve();
};
// The bootstrap subscription is kept ALIVE for the session — holding it open
// is the whole point; we resolve on the FIRST push (the initial State), which
// means the repo is now loaded in the session.
const unsub = subscribeDoc(nuri, () => done());
held.set(nuri, unsub);
// Bounded fallback: proceed even if no push arrives (a genuinely-absent doc
// reads 0 rows anyway — same as before — never a hang). NO polling.
setTimeout(done, OPEN_TIMEOUT_MS);
});
opened.add(nuri);
inFlight.delete(nuri);
})();
inFlight.set(nuri, p);
return p;
}
/**
* Open a SET of repos before an anchored batch read, in parallel, each tolerant
* ({@link ensureRepoOpen} never throws). Empty / falsy entries are ignored.
*/
export async function ensureReposOpen(nuris: Nuri[]): Promise<void> {
const unique = [...new Set(nuris.filter(Boolean))];
if (unique.length === 0) return;
await Promise.all(unique.map((n) => ensureRepoOpen(n)));
}
+15 -5
View File
@@ -43,6 +43,7 @@
import { docCreate, sparqlUpdate, sparqlQuery } from "./docs";
import { getCaps, getCurrentUser, getStoreRegistryDeps } from "./polyfill";
import { ensureReposOpen } from "./open-repo";
import { assertNuri } from "./sparql";
import type { Nuri } from "./types";
@@ -88,11 +89,13 @@ async function sessionId(): Promise<string> {
* This is O(1) in the doc's own size and independent of the rest of the (possibly
* bloated / shared) session store — it never iterates other graphs.
*
* All same-session repos (every doc `doc_create`d this session — in the mono-wallet
* polyfill, all of them) are already in `self.repos`, so the anchored query resolves
* the repo directly with no separate open. A genuinely-absent repo throws
* `RepoNotFound` here, in isolation, and the doc is skipped — never aborting the
* others (unlike the ORM fan-out). Returns the doc's rows, or `[]` on failure.
* COLD-START (fresh session, same persistent wallet): the repo is NOT in
* `self.repos` until something opens it, and an anchored query against an unopened
* repo silently returns 0 rows (never `RepoNotFound`). {@link readUnion} therefore
* opens the batch's repos ({@link ensureReposOpen}) BEFORE this read runs, so the
* anchored query resolves a same-session repo directly. A genuinely-absent repo
* still yields `[]` (in isolation, never aborting the others). Returns the doc's
* rows, or `[]` on failure.
*
* At the real multi-store migration this becomes a real sync: opening a per-user
* store repo by cap is a native broker fetch (`verifier.rs:1423` `OpenRepo` TODO).
@@ -137,6 +140,13 @@ export async function readUnion(docs: Nuri[]): Promise<UnionSubject[]> {
const unique = [...new Set(docs.filter(Boolean))];
if (unique.length === 0) return [];
// COLD-START heal (polyfill-era): on a fresh session over a persistent wallet the
// target repos are not yet in `self.repos`, so an anchored read would return 0
// rows. Open/subscribe each repo ONCE (idempotent, per session) and await its
// initial-state push before the anchored reads. No-op once opened / when the
// injected `ng` has no `doc_subscribe` (unit fake). See open-repo.ts.
await ensureReposOpen(unique);
// One anchored query per doc, in parallel, tolerant (a bad doc yields []).
const perDoc = await Promise.all(
unique.map(async (d) => ({ doc: assertNuri(d), rows: await readDoc(sid, d) })),
+9
View File
@@ -34,6 +34,7 @@
import { docCreate, sparqlUpdate, sparqlQuery } from "./docs";
import { getStoreRegistryDeps } from "./polyfill";
import { ensureRepoOpen } from "./open-repo";
import { escapeLiteral, escapeIri, assertNuri } from "./sparql";
import type { Nuri, Scope } from "./types";
@@ -444,6 +445,14 @@ export async function createEntityDoc(id: string, scope: Scope): Promise<Nuri> {
async function readScopeIndex(indexDoc: Nuri): Promise<Nuri[]> {
const s = await session();
const out: Nuri[] = [];
// COLD-START heal (polyfill-era): on a fresh session over a persistent wallet the
// scope-index repo (public OR protected — the protected one carries participations
// and is the one that most often reads empty) is not yet in `self.repos`, so this
// anchored read would return 0 NURIs → nothing gets listed → nothing gets
// subscribed (the self-inflicted circularity). Open/subscribe the index repo ONCE
// before reading it. Idempotent per session; no-op with the unit fake ng. See
// open-repo.ts.
await ensureRepoOpen(indexDoc);
try {
const res = await sparqlQuery(
s.sessionId,