refactor(client): open-repo attend le 1er STATE (barrière de sync), pas TabInfo
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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";
|
||||
|
||||
@@ -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<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>();
|
||||
/** 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<Nuri, SyncState>();
|
||||
/** 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<void> {
|
||||
/**
|
||||
* 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<void> {
|
||||
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<void>((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);
|
||||
|
||||
@@ -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: { <Variant>: … } }` 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<string, unknown>);
|
||||
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<string> {
|
||||
* 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 () => {};
|
||||
|
||||
Reference in New Issue
Block a user