Sylvain Duchesne
51b6278bf6
Its only callers were its own definition and its dedicated test block; no production code used it since the read-union removal (T01.f). The one test that exercised readDocumentSubjects' real integration path (not ensureReposOpen itself) is kept, moved under the ensureRepoOpen describe block it actually belongs to.
288 lines
17 KiB
TypeScript
288 lines
17 KiB
TypeScript
/**
|
|
* open-repo — cold-start repo opening for the ANCHORED read path (polyfill-era).
|
|
*
|
|
* ── The cold-start defect this heals ──────────────────────────────────────
|
|
* The anchored read path (`surface/read-document.ts` `readDocumentSubjects`,
|
|
* `shared-wallet/account-registry.ts` `readUserStore`) assumes the target repo is
|
|
* already usable by the verifier — 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): the repos are on the broker
|
|
* and in the profile's cache, but this session has not synced them, so a persisted
|
|
* document reads as empty.
|
|
*
|
|
* **The mechanism, corrected 2026-08-03.** This comment used to say the verifier
|
|
* "silently returns 0 rows (never a `RepoNotFound`)" for a repo absent from
|
|
* `self.repos`. That is FALSE at the source: `resolve_target_for_sparql` does
|
|
* `self.repos.get(repo_id).ok_or(NgError::RepoNotFound)?`
|
|
* (`engine/verifier/src/request_processor.rs:264,269`), which surfaces as a rejected
|
|
* promise. Two things produce the 0 rows actually observed, and neither is silence in
|
|
* the verifier: on a persistent profile `Verifier::load` repopulates `self.repos` from
|
|
* user storage at construction (`engine/verifier/src/verifier.rs:535-560`), so the repo
|
|
* is PRESENT but unsynced and the anchored query legitimately matches nothing; and this
|
|
* library's own per-document read caught every error and returned `[]`
|
|
* (`surface/read-model.ts`'s `readDoc`, until `1.0.0-dev.8` removed it along with the
|
|
* batch it served), so anything that did throw reached the caller as emptiness anyway.
|
|
* It no longer does: `surface/read-document.ts` lets the rejection through. The fix below is right; the diagnosis written beside it was not.
|
|
*
|
|
* 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 (`readUserStore`) 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. 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.
|
|
*
|
|
* ── 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 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
|
|
* 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 { 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 type { Nuri } from "../model/types";
|
|
|
|
/**
|
|
* 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".
|
|
*/
|
|
// @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>();
|
|
/** 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;
|
|
|
|
/**
|
|
* 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).
|
|
*/
|
|
let OPEN_TIMEOUT_MS = 8000;
|
|
|
|
/**
|
|
* Override the bootstrap-open fallback timeout (ms). TEST-ONLY: the timed-out
|
|
* branch (a doc that never pushes a `State`) is otherwise only reachable after the
|
|
* 8s production wait, too slow for a unit test. Production never calls this — the
|
|
* default stands. `resetOpenedRepos` restores the default.
|
|
*/
|
|
// @provenance setOpenTimeoutForTests kind=invention level=none ref=none — a test seam over the bounded fallback
|
|
export function setOpenTimeoutForTests(ms: number): void {
|
|
OPEN_TIMEOUT_MS = ms;
|
|
}
|
|
|
|
/** Reset the open registry (mainly for tests / a switched wallet). Tears down the
|
|
* held bootstrap subscriptions so a subsequent open re-subscribes cleanly. */
|
|
// @provenance resetOpenedRepos kind=invention level=none ref=none — per-session bookkeeping reset; upstream the verifier owns `self.repos` and nothing resets it from outside
|
|
export function resetOpenedRepos(): void {
|
|
for (const unsub of held.values()) {
|
|
try {
|
|
unsub();
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
opened.clear();
|
|
inFlight.clear();
|
|
held.clear();
|
|
syncState.clear();
|
|
boundSessionId = null;
|
|
OPEN_TIMEOUT_MS = 8000;
|
|
// Whatever ELSE was still subscribed to a document - an inbox observation, an
|
|
// application's own `subscribeDoc` - belongs to the session that just went away too. One real
|
|
// subscription now serves all of them (`surface/subscribe.ts`), so dropping the bootstrap
|
|
// ones above no longer closes those channels; they are re-opened against the new session,
|
|
// with their barrier forgotten. Without this a bootstrap open would join a surviving
|
|
// fan-out, be handed the PREVIOUS session's `State`, and call a repo the new session has
|
|
// never synced "synced". After the held teardown, so a document nobody else follows is
|
|
// simply released.
|
|
resubscribeDocs();
|
|
}
|
|
|
|
/**
|
|
* 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.
|
|
*/
|
|
// @provenance getSyncState kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — reports whether the first `State` arrived; upstream this is not observable from JS
|
|
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
|
|
* 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 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.
|
|
* Never throws; a failed open just leaves the read to behave as it did before.
|
|
*
|
|
* `holderKey` names WHOSE possession decides, when it is not the connected identity's —
|
|
* the emulated inbox processor opening an inbox repo for its owner during someone else's
|
|
* session (`emulated-verifier/inbox-processor.ts`). The public-store fetch is then skipped
|
|
* on purpose: it files what it obtains for whoever is CONNECTED, so asking on behalf of
|
|
* another holder would put a cap in the wrong ring, and a caller on that path has already
|
|
* established what its holder holds.
|
|
*/
|
|
// @provenance ensureRepoOpen kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — a real subscription held for the session — upstream `doc_subscribe` is what brings a repo into `self.repos`, which `resolve_target_for_sparql` then requires
|
|
export async function ensureRepoOpen(nuri: Nuri, holderKey?: string): Promise<void> {
|
|
if (!nuri) return;
|
|
// A repo in a PUBLIC store hands its cap to whoever asks — upstream by serving it on
|
|
// the outer overlay (`PublicRepoLinkV0`, `engine/net/src/types.rs:5098`). So ASK
|
|
// before deciding whether we may touch it, or the answer would be "no" purely for
|
|
// want of asking, and a bare reference to a public document would never suffice.
|
|
// Memoised and inert once the cap is held (see public-store.ts).
|
|
if (holderKey === undefined) await fetchReadCap(nuri);
|
|
// RULE 2 — do not even attempt. Opening a repo IS an access: it subscribes and
|
|
// pulls its state. A user that holds no cap for it has no business asking.
|
|
// (`ensurePhysicalRepoOpen` is the machinery's door — see physical.ts.)
|
|
if (mustNotAttempt(nuri, holderKey)) return;
|
|
return openRepoUnguarded(nuri);
|
|
}
|
|
|
|
/**
|
|
* The unguarded core. Exported for ONE importer — `shared-wallet/physical.ts` — and
|
|
* for nobody else; neither entry point re-exports it. The `Unguarded` suffix is the
|
|
* warning, and the single importer is what keeps it honest.
|
|
*/
|
|
// @provenance openRepoUnguarded kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — the same open without the reach guard; internal
|
|
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;
|
|
const pending = inFlight.get(nuri);
|
|
if (pending) return pending;
|
|
|
|
// 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");
|
|
// Barrier clock: how long the subscribe→first-State (or fallback) round-trip
|
|
// took, surfaced on the BARRIER trace line below — the most important line in
|
|
// the whole low-level data-path trace: it distinguishes a genuine absence
|
|
// (`synced` → a 0-row read means it) from a not-yet-synced read (`timed-out`).
|
|
const barrierStartedAt = Date.now();
|
|
|
|
const p = (async () => {
|
|
await new Promise<void>((resolve) => {
|
|
let settled = false;
|
|
// Resolve on the first `State` (the sync BARRIER), marking the nuri "synced".
|
|
const onState = (): void => {
|
|
if (settled) return;
|
|
settled = true;
|
|
syncState.set(nuri, "synced");
|
|
logStage("BARRIER " + shortNuri(nuri) + " synced (" + (Date.now() - barrierStartedAt) + "ms)");
|
|
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");
|
|
logStage("BARRIER " + shortNuri(nuri) + " timed-out (" + (Date.now() - barrierStartedAt) + "ms)");
|
|
resolve();
|
|
};
|
|
// The bootstrap subscription is kept ALIVE for the session — holding it open
|
|
// 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.
|
|
// 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();
|
|
});
|
|
held.set(nuri, unsub);
|
|
setTimeout(onTimeout, OPEN_TIMEOUT_MS);
|
|
});
|
|
opened.add(nuri);
|
|
inFlight.delete(nuri);
|
|
})();
|
|
inFlight.set(nuri, p);
|
|
return p;
|
|
}
|