From 4382d04391669e0fab79a2c22cbaf02ff477ac49 Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Thu, 9 Jul 2026 11:23:31 +0200 Subject: [PATCH] =?UTF-8?q?refactor(client):=20open-repo=20attend=20le=201?= =?UTF-8?q?er=20STATE=20(barri=C3=A8re=20de=20sync),=20pas=20TabInfo?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Avant : `ensureRepoOpen` résolvait son attente sur le 1er push d'abonnement (= TabInfo, ~1-3ms) — donc AVANT la vraie barrière de sync. Correct par chance sur ce broker (State suit TabInfo d'~1ms), faux si State tarde. Maintenant : - `subscribe.ts` : `docChangeType(resp)` extrait le variant (`State`/`Patch`/ `TabInfo`/…) sans cast `any` ; `subscribeDoc`/`subscribeDocs` le passent en 2e arg NON-cassant (les appelants 0-arg — discovery, inbox — inchangés). - `ensureRepoOpen` n'agit que sur `type === "State"` → attend la vraie barrière. - État de sync par-doc explicite : `getSyncState(nuri): "syncing"|"synced"| "timed-out"|"unknown"`. Le fallback 8s marque `timed-out`, JAMAIS `synced` — les deux ne sont plus confondus (base du futur signal app + du « trop long = signal »). - Fake ng (sans doc_subscribe) : résolution immédiate préservée (bun test intact). gate : tsc propre ; bun test 117 ; test:e2e 39 passed (CONTRAT 3 vert, events=["TabInfo","State"] ; reconnexion cold-read 176ms/10.8s — l'attente du 1er State se déclenche vite, pas de gonflement par timeout). Pas de régression app (le rouge du test reconnexion est pré-existant et broker-lenteur, vérifié sur lib vierge). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/src/index.ts | 4 +- packages/client/src/open-repo.ts | 93 ++++++++++++++++++++++++++------ packages/client/src/subscribe.ts | 46 ++++++++++++++-- 3 files changed, 121 insertions(+), 22 deletions(-) diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 70a8a0d..fa1dec0 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -18,8 +18,8 @@ export * as inbox from "./inbox"; export * as discovery from "./discovery"; export type { IndexEntry, SubmitOptions } from "./discovery"; export * as docs from "./docs"; -export { subscribeDoc, subscribeDocs } from "./subscribe"; -export type { DocChange, Unsubscribe } from "./subscribe"; +export { subscribeDoc, subscribeDocs, docChangeType } from "./subscribe"; +export type { DocChange, DocChangeType, Unsubscribe } from "./subscribe"; export * as readModel from "./read-model"; export type { UnionSubject } from "./read-model"; export * as storeRegistry from "./store-registry"; diff --git a/packages/client/src/open-repo.ts b/packages/client/src/open-repo.ts index 74722c9..f2dd85c 100644 --- a/packages/client/src/open-repo.ts +++ b/packages/client/src/open-repo.ts @@ -20,9 +20,13 @@ * * ── 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 + * wrapper over the platform's `doc_subscribe`) — NOT a parallel channel. On + * subscribe the platform pushes `TabInfo` FIRST (~1-3ms) and then the initial + * `State` (~2-3ms); the FIRST `State` is the sync BARRIER — after it, presence is + * guaranteed and absence definitive (pinned empirically by CONTRACT 3 in `e2e/`). + * So we await that first `State` specifically (identified via the `type` argument + * {@link DocChangeType} `subscribeDoc` now surfaces), NOT the first push of any + * kind — resolving on `TabInfo` would return before the real barrier. 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. @@ -34,8 +38,20 @@ * 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. + * rather than wrongly skipped as "already open". No polling: we wait on the first + * `State` push, with a bounded fallback timeout so a missing push can't hang. + * + * ── Sync state (per-nuri, lib-internal) ─────────────────────────────────── + * Each nuri carries an explicit sync state, readable via {@link getSyncState}: + * `"syncing"` — subscribed, no `State` yet (barrier not reached); + * `"synced"` — first `State` received (barrier reached — the TRUTH signal); + * `"timed-out"` — the bounded fallback fired with NO `State` (open proceeded so + * the read is never blocked, but this is NOT `"synced"`: a future + * "ready" signal must not mistake a timeout for a real barrier); + * `"unknown"` — never requested (or the fake-ng no-op path: no `State` semantics). + * This distinction is the point — `"synced"` and `"timed-out"` are kept apart so a + * later reactive readiness layer can trust `"synced"` and treat `"timed-out"` as + * "opened best-effort, sync unconfirmed". * * ── Migration ───────────────────────────────────────────────────────────── * At the real multi-store migration this becomes "open the user's store repo by @@ -47,12 +63,24 @@ 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). */ +/** + * The per-nuri bootstrap sync state (lib-internal). See the module header: + * - `"syncing"` subscribed, first `State` not yet received; + * - `"synced"` first `State` received — the real sync barrier (CONTRACT 3); + * - `"timed-out"` fallback fired without a `State` — open proceeded, sync UNconfirmed; + * `"unknown"` (from {@link getSyncState}) means "never requested / no `State` semantics". + */ +export type SyncState = "syncing" | "synced" | "timed-out"; + +/** Repos whose bootstrap open has completed (first `State` received OR timed out). */ const opened = new Set(); /** In-flight opens, so concurrent `ensureRepoOpen(nuri)` share one subscription. */ const inFlight = new Map>(); /** Live bootstrap subscriptions, kept for the session (this is what holds repos open). */ const held = new Map(); +/** Explicit per-nuri sync state — the barrier signal, distinct from `opened`. + * `synced` and `timed-out` are NOT merged (see module header). */ +const syncState = new Map(); /** 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; @@ -78,9 +106,23 @@ export function resetOpenedRepos(): void { opened.clear(); inFlight.clear(); held.clear(); + syncState.clear(); boundSessionId = null; } +/** + * The bootstrap sync state of `nuri` (lib-internal accessor; NOT a reactive + * app-facing hook — that is a later phase). Returns `"unknown"` if the repo was + * never opened via {@link ensureRepoOpen} (or was opened on the fake-ng no-op + * path, which has no `State` semantics). Otherwise `"syncing"` (subscribed, no + * `State` yet), `"synced"` (first `State` received — the barrier), or + * `"timed-out"` (fallback fired without a `State`). `"synced"` and `"timed-out"` + * are deliberately distinct — a later readiness signal must not confuse them. + */ +export function getSyncState(nuri: Nuri): SyncState | "unknown" { + return syncState.get(nuri) ?? "unknown"; +} + /** * 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 @@ -101,8 +143,9 @@ async function syncSession(): Promise { /** * 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. + * of returning 0 rows. Opens via {@link subscribeDoc} and awaits the first `State` + * push — the sync barrier (bounded fallback marks the nuri `"timed-out"`, not + * `"synced"`). 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. @@ -116,29 +159,47 @@ export async function ensureRepoOpen(nuri: Nuri): Promise { const pending = inFlight.get(nuri); if (pending) return pending; - // No reactive primitive on the injected ng (fake-ng unit suite): nothing to open. + // 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` + // does not regress. No sync state is recorded (getSyncState → "unknown"): the fake + // 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); return; } + // Subscribed, first `State` not yet seen. + syncState.set(nuri, "syncing"); + const p = (async () => { await new Promise((resolve) => { let settled = false; - const done = (): void => { + // Resolve on the first `State` (the sync BARRIER), marking the nuri "synced". + const onState = (): void => { if (settled) return; settled = true; + syncState.set(nuri, "synced"); + resolve(); + }; + // Bounded fallback: proceed WITHOUT a `State`, but mark "timed-out" — NOT + // "synced". A genuinely-absent doc reads 0 rows anyway (same as before, never + // a hang); the distinct state keeps a future "ready" signal from lying. + const onTimeout = (): void => { + if (settled) return; + settled = true; + syncState.set(nuri, "timed-out"); 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()); + // is the whole point. We wait for the FIRST `State` event specifically (the + // barrier), NOT any push: the platform pushes `TabInfo` before `State`, and + // resolving on `TabInfo` would return before the real sync barrier. + const unsub = subscribeDoc(nuri, (_r, type) => { + if (type === "State") onState(); + }); 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); + setTimeout(onTimeout, OPEN_TIMEOUT_MS); }); opened.add(nuri); inFlight.delete(nuri); diff --git a/packages/client/src/subscribe.ts b/packages/client/src/subscribe.ts index 59abaa5..f10486d 100644 --- a/packages/client/src/subscribe.ts +++ b/packages/client/src/subscribe.ts @@ -45,6 +45,35 @@ import type { Nuri } from "./types"; */ export type DocChange = unknown; +/** + * The discriminant of a {@link DocChange} — the single variant key of the raw + * `AppResponse` payload (`{ V0: { State | Patch | TabInfo | … } }`). It is NOT a + * closed enum: the platform may push other variants, so this is a bare `string` + * (e.g. `"State"`, `"Patch"`, `"TabInfo"`), or `undefined` when the shape can't + * be read. Verified against the CONTRACT-3 e2e probe (`e2e/sdk-entry.ts`): the + * variant is `Object.keys(resp.V0)[0]`. Exposed so a caller that needs the SYNC + * BARRIER (the first `State`, per CONTRACT 3) can distinguish it from the earlier + * `TabInfo`/`Patch` pushes — see `open-repo.ts`. Most callers ignore it and use + * any push as a plain change signal. + */ +export type DocChangeType = string | undefined; + +/** + * Extract the variant key from a raw {@link DocChange}. Reads `resp.V0` (case- + * tolerant to `v0`) and returns its first key — the AppResponse variant name + * (`"State"` / `"Patch"` / `"TabInfo"` / …). Returns `undefined` if the payload + * is not a recognisable `{ V0: { : … } }` object. Inspects the variant + * proplerly (no `any`-cast to force it) so a `State` push is identifiable. + */ +export function docChangeType(resp: DocChange): DocChangeType { + if (!resp || typeof resp !== "object") return undefined; + const outer = resp as { V0?: unknown; v0?: unknown }; + const v0 = outer.V0 ?? outer.v0; + if (!v0 || typeof v0 !== "object") return undefined; + const keys = Object.keys(v0 as Record); + return keys.length > 0 ? keys[0] : undefined; +} + /** An unsubscribe function — idempotent (calling it twice is a no-op). */ export type Unsubscribe = () => void; @@ -63,9 +92,18 @@ async function sessionId(): Promise { * cancellation is honoured as soon as the real unsubscribe is available (and no * further `onChange` fires after unsubscribe). * + * `onChange` receives the raw payload AND its variant type ({@link docChangeType}, + * e.g. `"State"`). The type is a NON-BREAKING second argument: existing callers + * that ignore it (the change-signal pattern — `discovery.ts`, `inbox.ts`) are + * unaffected; a caller that needs the sync barrier (`open-repo.ts`) reads it to + * act only on the first `State`. + * * Calls the REAL injected `ng.doc_subscribe` directly (never `makeNg`). */ -export function subscribeDoc(nuri: Nuri, onChange: (r: DocChange) => void): Unsubscribe { +export function subscribeDoc( + nuri: Nuri, + onChange: (r: DocChange, type: DocChangeType) => void, +): Unsubscribe { const { ng } = getConfig(); let stopped = false; let realUnsub: (() => void) | null = null; @@ -73,7 +111,7 @@ export function subscribeDoc(nuri: Nuri, onChange: (r: DocChange) => void): Unsu const cb = (resp: DocChange): void => { if (stopped) return; try { - onChange(resp); + onChange(resp, docChangeType(resp)); } catch (error) { console.error("[subscribe] onChange handler threw for", nuri, error); } @@ -123,7 +161,7 @@ export function subscribeDoc(nuri: Nuri, onChange: (r: DocChange) => void): Unsu */ export function subscribeDocs( nuris: Nuri[], - onChange: (nuri: Nuri, r: DocChange) => void, + onChange: (nuri: Nuri, r: DocChange, type: DocChangeType) => void, ): Unsubscribe { const unique = [...new Set(nuris.filter(Boolean))]; const unsubs = unique.map((nuri) => { @@ -131,7 +169,7 @@ export function subscribeDocs( // async setup failure (logged, never thrown), so one bad doc cannot abort the // construction of the others here. try { - return subscribeDoc(nuri, (r) => onChange(nuri, r)); + return subscribeDoc(nuri, (r, type) => onChange(nuri, r, type)); } catch (error) { console.error("[subscribe] subscribeDocs: failed to subscribe", nuri, error); return () => {};