fix(client): dé-dupliquer les ensureAccount concurrents au cold-start (fork résolu)

Dernière couche du bug de reconnexion : au cold-start, `watchShape` public +
protected + l'effet owned-events appellent `ensureAccount(A)` quasi-simultanément
AVANT la sync du shim → chacun lit 0 → chacun provisionne un nouveau jeu de docs
(fork par-appelant) → la résolution déterministe canonique fait alors diverger
lecteur et écrivain sur le docProtected → `readScopeIndex` vide.

Fix : `ensureInFlight` (map de promesses) dé-duplique les provisions concurrentes
en UNE seule ; `discovery.readIndex` ouvre son repo au cold-start (`ensureRepoOpen`).
Avec la résolution canonique déjà committée, écrivain et lecteur convergent.
Mesuré (levier isSuccess) : la participation protected converge `isSuccess=true,
data=1` sur la page fraîche (plus « vide à 30s »).

gate : tsc 0 ; bun test 123 ; test:e2e 42/42.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sylvain Duchesne
2026-07-10 11:29:59 +02:00
parent bd48b16e31
commit 9103996dbe
4 changed files with 127 additions and 40 deletions
+14
View File
@@ -39,6 +39,7 @@
import * as inbox from "./inbox"; import * as inbox from "./inbox";
import { subscribeDoc } from "./subscribe"; import { subscribeDoc } from "./subscribe";
import { ensureRepoOpen } from "./open-repo";
import { ensureAccount, reservedAccount } from "./store-registry"; import { ensureAccount, reservedAccount } from "./store-registry";
import { getCaps } from "./polyfill"; import { getCaps } from "./polyfill";
import type { Nuri, PrincipalId } from "./types"; import type { Nuri, PrincipalId } from "./types";
@@ -154,6 +155,19 @@ export async function submitToIndex(ref: unknown, opts?: SubmitOptions): Promise
*/ */
export async function readIndex(): Promise<IndexEntry[]> { export async function readIndex(): Promise<IndexEntry[]> {
const target = await indexInboxNuri(); const target = await indexInboxNuri();
// COLD-START heal (polyfill-era): on a FRESH session over a persistent wallet the
// discovery-index inbox repo is not yet in the verifier's `self.repos`, so the
// anchored `inbox.read` below would resolve an unopened repo and silently return 0
// deposits — the same self-inflicted cold-read gap `readScopeIndex`/`readUnion`
// heal. This is what made the PUBLIC read's discovery fold come back empty on a
// reconnect, so a fresh page's home stayed empty for tens of seconds while the doc
// slowly synced by other means. Open/subscribe the index repo ONCE and await its
// first `State` (the sync barrier) before the anchored read. Idempotent per session;
// no-op with the unit fake ng (no `doc_subscribe`). This is done HERE (a cold direct
// reader) rather than inside `inbox.read`, because `inbox.watch` already holds the
// repo open via its own subscription and must not spawn a second bootstrap open. See
// open-repo.ts.
await ensureRepoOpen(target);
const deposits = await inbox.read(target); const deposits = await inbox.read(target);
const seen = new Set<string>(); const seen = new Set<string>();
const entries: IndexEntry[] = []; const entries: IndexEntry[] = [];
+7
View File
@@ -156,6 +156,13 @@ export async function post(targetInbox: Nuri, opts: PostOptions): Promise<void>
*/ */
export async function read(targetInbox: Nuri): Promise<Deposit[]> { export async function read(targetInbox: Nuri): Promise<Deposit[]> {
const sid = await sessionId(); const sid = await sessionId();
// NOTE: cold-start repo opening is done by the COLD DIRECT readers that need it
// (e.g. `discovery.readIndex` → `ensureInboxRepoOpen`), NOT here — `inbox.watch`
// already holds the repo open via its own `subscribeDoc`, so opening a second
// bootstrap subscription from inside a watch's re-read would be redundant and can
// race the watch's own initial-`State` delivery. Keeping `read` a pure anchored
// read leaves both callers correct: the watch path stays event-driven, and the
// cold direct-read path opens the repo explicitly before calling `read`.
// NO explicit `GRAPH <…>` clause — read the anchored DEFAULT graph (see the // NO explicit `GRAPH <…>` clause — read the anchored DEFAULT graph (see the
// note in `post`). The anchor (`targetInbox`) scopes the query to that repo's // note in `post`). The anchor (`targetInbox`) scopes the query to that repo's
// default graph, exactly where `post` writes. // default graph, exactly where `post` writes.
+45
View File
@@ -381,12 +381,49 @@ async function createDoc(): Promise<Nuri> {
return docCreate(s.sessionId, "Graph", "data:graph", "store", undefined); return docCreate(s.sessionId, "Graph", "data:graph", "store", undefined);
} }
/**
* In-flight `ensureAccount` promises, keyed by account key — so CONCURRENT
* `ensureAccount(id)` calls for the SAME account share ONE resolve-or-provision.
*
* ── Why this is load-bearing (the reconnection FORK root cause) ─────────────
* On a fresh page over the persistent wallet, SEVERAL independent callers hit
* `ensureAccount(A)` near-simultaneously: `watchShape('public')` and
* `watchShape('protected')` each resolve their doc set (`listMyEntityDocs` →
* `ensureAccount`) and subscribe their container (`scopeIndexDoc` →
* `ensureAccount`), plus the app's owned-events effect. Each runs BEFORE the shim
* has synced, so each `resolveAccountReliably` reads 0 rows and, independently,
* PROVISIONS a brand-new set of scope docs — an account FORK PER caller. Measured:
* a single new account accrued EIGHT distinct `docPublic`/`docProtected` values in
* one session. `canonicalDoc` then makes a fresh reader pick the lexicographic-min
* doc, which differs from the (cached, freshly-created) doc the WRITER wrote into →
* the reader lists an empty scope index → the home/participation reads come back
* empty (the reconnection bug's last layer).
*
* De-duping concurrent provisions collapses those N racing provisions into ONE:
* the first caller resolves-or-provisions; every concurrent caller awaits the SAME
* promise and gets the SAME record, so no duplicate docs are ever minted and writer
* and reader converge on one canonical scope doc. Not polling: a bounded in-memory
* promise map (mirrors open-repo.ts `inFlight`), cleared the instant it settles.
*/
const ensureInFlight = new Map<string, Promise<AccountRecord>>();
/** /**
* Ensure an account exists in the shim, creating its 3 scope documents on * Ensure an account exists in the shim, creating its 3 scope documents on
* first sight. Idempotent — returns the existing record if already present. * first sight. Idempotent — returns the existing record if already present.
* Concurrency-safe: concurrent calls for the same account share one provision
* (see {@link ensureInFlight}) so a fresh page never FORKS the account.
*/ */
export async function ensureAccount(id: string): Promise<AccountRecord> { export async function ensureAccount(id: string): Promise<AccountRecord> {
const key = accountKey(id); const key = accountKey(id);
// A completed provision/resolve is cached → no query, no fork risk.
const cached = accountCache.get(key);
if (cached) return cached;
// A concurrent provision for the SAME account is already running → await it,
// instead of racing a second (forking) provision. This is the anti-fork guard.
const pending = ensureInFlight.get(key);
if (pending) return pending;
const p = (async (): Promise<AccountRecord> => {
// HOT PATH: targeted O(1) lookup — does THIS account already exist? — instead // HOT PATH: targeted O(1) lookup — does THIS account already exist? — instead
// of a full-shim scan (loadShim). Off the read/write hot path entirely. // of a full-shim scan (loadShim). Off the read/write hot path entirely.
// //
@@ -431,6 +468,14 @@ export async function ensureAccount(id: string): Promise<AccountRecord> {
accountCache.set(key, record); accountCache.set(key, record);
cache?.set(key, record); cache?.set(key, record);
return record; return record;
})();
ensureInFlight.set(key, p);
try {
return await p;
} finally {
ensureInFlight.delete(key);
}
} }
// --- resolvers ------------------------------------------------------------ // --- resolvers ------------------------------------------------------------
@@ -188,6 +188,27 @@ test("ensureAccount is idempotent (case/@-insensitive key), no extra docs", asyn
expect(fake.doc_create).toHaveBeenCalledTimes(3); // not 6 expect(fake.doc_create).toHaveBeenCalledTimes(3); // not 6
}); });
test("ensureAccount de-dupes CONCURRENT provisions (anti-fork): one account, 3 docs", async () => {
// The reconnection FORK: several callers (watchShape public+protected, the
// container subs, the owned-events effect) hit ensureAccount(SAME id) BEFORE
// the shim has synced, so each reads 0 rows and independently provisions a new
// set of scope docs — N forks, N×3 docs, duplicate docPublic/docProtected in the
// shim → a fresh reader picks a different canonical doc than the writer wrote to.
// The in-flight de-dup collapses N concurrent provisions into ONE.
const results = await Promise.all([
ensureAccount("Bob"),
ensureAccount("Bob"),
ensureAccount("@bob"),
ensureAccount("BOB"),
ensureAccount("bob"),
]);
// Exactly ONE set of 3 docs was created — not 5×3.
expect(fake.doc_create).toHaveBeenCalledTimes(3);
// Every caller got the SAME record (same docs), so writer/reader can never
// disagree on the canonical scope doc.
for (const r of results) expect(r).toEqual(results[0]!);
});
test("loadShim round-trips a persisted account across a cache reset", async () => { test("loadShim round-trips a persisted account across a cache reset", async () => {
await ensureAccount("Bob"); await ensureAccount("Bob");
resetRegistryCache(); // force a re-read from the fake store resetRegistryCache(); // force a re-read from the fake store