Reconnexion/ng-eventually
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat!: une seule reponse a « ce document est-il ouvert dans cette session »

Browse Source 
Trois reponses coexistaient et pouvaient se contredire : un ensemble de
souscriptions bootstrap et une carte d'etat de barriere dans `open-repo`, et les
auditeurs du fan-out dans `surface/subscribe`. Les deux premieres survivaient a
la disparition de la troisieme, parce que `resetDocSubscriptions()` - ce que fait
le `configure()` publie - largue ses auditeurs sans rien dire a personne.

Il n'en reste qu'une, l'etat de l'unique souscription reelle du document, exposee
par `docChannel(nuri)`. `open-repo` n'enregistre plus que l'auditeur qu'il a pose
- « est-ce que j'ecoute encore », rien de plus - et derive `getSyncState` a chaque
lecture, si bien qu'aucune reponse ne peut survivre au canal qu'elle decrit.
Disparaissent : `held`, `syncState`, `inFlight`, `settleInFlight` et le compteur
de generation, remplace par l'identite de l'enregistrement.

Les deux echecs prouves deviennent des tests qui echouent sans le changement :
un second `configure()` apres une synchro reussie (`"synced"` au lieu de
`"timed-out"`, zero re-souscription, et `isSuccess` cote application), et un
etablissement encore en cours quand la couche de souscription est demontee - cas
que le canal `onSetupFailed` ne peut pas rattraper, puisqu'il ne rapporte que la
tentative encore attendue.

La ligne qui ne casse pas : un document synchronise qui ne contient rien repond
toujours un succes a zero objet. `1.0.0-dev.10`, les deux non-garanties
nommees sont retirees et la garantie livree porte le test qui l'epingle.
This commit is contained in:
Sylvain Duchesne
2026-09-24 12:57:43 +02:00
parent 2da26c5c4d
commit de9b8a0d67
10 changed files with 574 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/polyfill/src/emulated-verifier/open-repo.ts
+226 -169
View File
@@ -45,26 +45,31 @@
* 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.
* A repo is opened once per session, and what makes that safe is ASKING rather than
* remembering: {@link opens} records the bootstrap listener this module took out, and
* {@link docChannel} says whether the subscription behind it stands. A caller skips the work
* only when both hold - see the invariant below, and `openRepoUnguarded`, where the two
* halves are named. Concurrent callers join one wait ({@link BootstrapOpen.done}). A
* brand-new page / module instance starts empty; 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".
* ── Sync state (per-nuri, lib-internal, DERIVED) ──────────────────────────
* {@link getSyncState} computes a nuri's barrier state; nothing stores it:
* `"syncing"` - this module is inside its bounded wait, no `State` yet;
* `"synced"` - a `State` has arrived on the channel standing NOW (barrier reached,
* the TRUTH signal: a zero-row read over it MEANS zero);
* `"timed-out"` - opened here, and no barrier to show for it: the wait gave up, the
* subscribe was refused, or the channel was taken away. The read is never
* blocked, but this is NOT `"synced"` - a readiness signal must not
* mistake either for a real barrier;
* `"unknown"` - never opened here (or the fake-ng no-op path: no `State` semantics).
* `"synced"` and `"timed-out"` are kept apart for that reason, and `"unknown"` is the one
* value that is deliberately treated as authoritative, because where there are no `State`
* semantics there is no barrier to have missed.
*
* ── Migration ─────────────────────────────────────────────────────────────
* At the real multi-store migration this becomes "open the user's store repo by
@@ -75,69 +80,104 @@
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 { docChannel, resubscribeDocs, subscribeDocUnguarded, type Unsubscribe } from "../surface/subscribe";
import { accessLogPrefix, 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".
* The per-nuri bootstrap sync state (lib-internal), DERIVED on every read by
* {@link getSyncState} - nothing stores it. See the module header:
* - `"syncing"` inside the bounded wait, first `State` not yet received;
* - `"synced"` a `State` arrived on the channel standing NOW - the real sync barrier
* (CONTRACT 3);
* - `"timed-out"` opened here with no barrier to show for it - the wait gave up, the
* subscribe was refused, or the channel was taken away. Sync UNconfirmed;
* `"unknown"` (from {@link getSyncState}) means "never opened here / 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";
/*
* WHAT THIS REGISTRY MEANS, and it is the thing that went wrong five rounds running.
* THE INVARIANT, and it is the thing that went wrong six rounds running.
*
* `held` is the ONE record of openness: **this session holds a live bootstrap subscription
* to this document**. Nothing else records it, because two records of the same fact can
* disagree and this pair did. There was an `opened` set beside it meaning "we once called
* `doc_subscribe` and stopped waiting", and that looser meaning is what lied three ways:
* it survived a subscribe the broker REJECTED (so the retry `subscribe.ts` offers was
* blocked for ever), it survived a registry reset (so a new session short-circuited on the
* previous one's record), and it let a document report a barrier it had never crossed.
* **"Is this document open in this session" is answered in ONE place - the state of the
* document's one real subscription (`surface/subscribe.ts`, {@link docChannel}) - and this
* module records only the bootstrap listener it TOOK OUT, never whether that listener ever
* became a live channel or crossed a barrier.** So no answer here can outlive the channel
* it describes, because none is kept.
*
* The three consequences, each now true by construction:
* - a subscribe that FAILED records nothing, so the next caller subscribes again;
* - a reset clears the record AND downgrades the sync state, so nothing claims authority
* the new session has not earned;
* - `syncState` says `"unknown"` for one reason only: no barrier was ever expected here
* (never requested, or an `ng` with no `doc_subscribe` and so no `State` semantics).
* It never means "we forgot".
* What that replaces, and why each had to go rather than be repaired:
* - a `held` set of "live bootstrap subscriptions". It was written the moment
* `subscribeDocUnguarded` returned, which is BEFORE the `doc_subscribe` behind it has
* been placed - so it meant "we asked", the looser meaning an `opened` set had been
* deleted for one round earlier. And `resetDocSubscriptions` (which PUBLISHED
* `configure()` calls) drops every listener without a word, so it also survived the
* teardown of the very subscription it named.
* - a `syncState` map of barrier states. Same fate, one step worse: it kept saying
* `"synced"` for a document whose channel had gone, and `"synced"` is AUTHORITY - a
* zero-row read over it is a definitive absence. A read that had been correctly
* withheld came back a success, for a document this session had never subscribed.
*
* Both facts live in the fan-out already and are forgotten there at exactly the right
* moments, which is what makes asking better than remembering: a channel released, a
* session changed under it, an establish refused.
*
* {@link getSyncState}'s four values are therefore DERIVED, and `"unknown"` still means one
* thing only: no barrier was ever expected here (never requested, or an `ng` with no
* `doc_subscribe` and so no `State` semantics). It never means "we forgot".
*/
/** In-flight opens... see {@link inFlight} below; `held` is the record of openness. */
/** In-flight opens, so concurrent `ensureRepoOpen(nuri)` share one subscription. */
const inFlight = new Map<Nuri, Promise<void>>();
/**
* How to END an in-flight open from outside it - see {@link resetOpenedRepos}, the only
* caller. An open normally settles on the first `State` or on its own bounded fallback,
* and both of those live inside the subscription; a reset tears that subscription down,
* so without this the open is left with nothing that can ever finish it but a timer whose
* whole purpose has gone.
* The bootstrap open this module ASKED FOR - never a record of whether it succeeded.
*
* One per document, and its presence is the whole of what this module knows: *`open-repo`
* has opened this document in this session*. Everything else about it is a question for
* {@link docChannel}.
*/
const settleInFlight = new Map<Nuri, () => 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
interface BootstrapOpen {
/**
* The bounded wait this open exposes to its callers, resolved on the first `State`, on the
* fallback, on a refused subscribe, or by a reset. Concurrent callers join it rather than
* opening a second time - which is what an `inFlight` map used to do.
*/
done: Promise<void>;
/**
* Release the bootstrap listener this open took out. `null` once it has been released -
* by a refused subscribe, or by {@link resetOpenedRepos} - and that is ALL it says: it
* answers "am I still listening", never "is this document open". The second question needs
* {@link docChannel} as well, because a listener can be dropped from under this module
* (`resetDocSubscriptions`) and nothing here would be told.
*/
release: Unsubscribe | null;
/**
* How to END the wait from outside it - {@link resetOpenedRepos}, the only caller; `null`
* once the wait is over. An open normally settles on the first `State` or on its own
* bounded fallback, and both of those live inside the subscription; a reset tears that
* subscription down, so without this the open is left with nothing that can ever finish it
* but a timer whose whole purpose has gone.
*/
settle: (() => void) | null;
/** Inside the bounded wait: no `State` yet, and the fallback has not fired. */
waiting: boolean;
}
/** Every document this module has opened in this session. See {@link BootstrapOpen}. */
const opens = new Map<Nuri, BootstrapOpen>();
/**
* A document this module opened and no longer holds anything for - what a reset leaves
* behind. It is deliberately NOT a deletion: forgetting it entirely would make
* {@link getSyncState} answer `"unknown"`, which means "no barrier was ever expected here"
* and is therefore read as authoritative. A document whose subscription a reset took away is
* the opposite of that.
*
* It is also the generation guard. A reset REPLACES the record rather than mutating it, so
* an open still running against the previous one compares `opens.get(nuri) !== mine` and
* writes nothing - object identity doing what a counter used to.
*/
function disowned(): BootstrapOpen {
return { done: Promise.resolve(), release: null, settle: null, waiting: false };
}
/** The session id the current {@link opens} entries belong to. A change means a
* new verifier session (fresh `self.repos`) → the registry must be invalidated. */
let boundSessionId: string | number | null = null;
/**
* Which GENERATION of the registry an open belongs to, bumped by {@link resetOpenedRepos}.
*
* An open's continuation records the repo as opened, and it runs one microtask after the
* open settles - a window a reset can land in. Without this it wrote that record into the
* registry that had just replaced its own, and every later open of the same nuri
* short-circuited on it: the repo was never subscribed in the new session, and a read over
* it answered as though it had been. A reset must INVALIDATE what it abandons, not only
* end it.
*/
let registryEpoch = 0;
/**
* Max wait (ms) for the initial-state push before proceeding with the read anyway.
@@ -162,40 +202,30 @@ export function setOpenTimeoutForTests(ms: number): void {
* 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()) {
for (const [nuri, open] of [...opens]) {
// Let the bootstrap listener go first: it is what held the repo open for the session
// that is ending, and a document nobody else follows is simply released.
try {
unsub();
open.release?.();
} catch {
/* ignore */
}
// END the open this reset abandons: a caller is already awaiting it, and the `State` it
// was waiting for cannot arrive now that the subscription has been torn down. It used to
// wait out the full fallback - eight seconds of a read going nowhere. Taken before the
// record is replaced, and called after, so the waiter's continuation finds the record
// already gone and writes nothing.
const settle = open.settle;
open.settle = null;
// DISOWNED rather than forgotten - see {@link disowned}. The barrier needs no downgrading
// of its own any more: it was never recorded here, and the fan-out forgets it on exactly
// these events.
opens.set(nuri, disowned());
// The wait's own end, not a shortcut past it: clearing `waiting` here instead would make
// this call a no-op and leave the caller hanging on a subscription that no longer exists -
// the eight-second read the abandoned-open fix exists to prevent.
settle?.();
}
// END the opens this reset abandons, BEFORE the maps go: a caller is already awaiting
// one, and the `State` it was waiting for cannot arrive now that the subscription above
// has been torn down. It used to wait out the full fallback - eight seconds of a read
// going nowhere - and then proceed with its sync state CLEARED, which reads as
// `"unknown"`; and `unknown` is the LAX answer (`emulated-verifier/index-deposit.ts`
// `noRowsMeansAbsent` accepts it), so a zero-row read taken right after a session change
// could be called a definitive absence. Marked `"timed-out"` instead, which is the exact
// truth - opened best-effort, sync unconfirmed - and the reading that refuses to call
// nothing an absence. The abandoned open's own fallback fires later and sets the same
// value, so the two cannot disagree.
const abandoned = [...settleInFlight.entries()];
// Every document this reset takes a subscription away from, in flight or long settled.
// Their sync state is DOWNGRADED rather than cleared: a barrier crossed in the session
// that just went away is not one this session has crossed, and clearing would leave them
// reading `"unknown"` - which means "no barrier was ever expected here" and is therefore
// treated as authoritative. That conflation is how a read that had been correctly
// withheld came back claiming a success, against a query nothing had re-subscribed.
const disowned = new Set<Nuri>([...held.keys(), ...abandoned.map(([nuri]) => nuri)]);
// Everything in flight now belongs to a registry that no longer exists, so its
// continuation must write nothing - see {@link registryEpoch}.
registryEpoch += 1;
inFlight.clear();
settleInFlight.clear();
held.clear();
syncState.clear();
for (const nuri of disowned) syncState.set(nuri, "timed-out");
for (const [, settle] of abandoned) settle();
boundSessionId = null;
OPEN_TIMEOUT_MS = 8000;
// Whatever ELSE was still subscribed to a document - an inbox observation, an
@@ -204,8 +234,8 @@ export function resetOpenedRepos(): void {
// 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.
// never synced "synced". After the listeners above are released, so a document nobody else
// follows is simply released.
resubscribeDocs();
}
@@ -220,7 +250,22 @@ export function resetOpenedRepos(): void {
*/
// @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";
const channel = docChannel(nuri);
// The first `State` on the channel standing NOW. Asked first and asked of the fan-out,
// because that is the one record of it: a barrier is forgotten when its channel is released
// or its session changes, so this cannot report one that has gone. Answered even for a
// document this module never opened - an application's own subscription crosses the same
// barrier, and `"synced"` is then simply true.
if (channel.barrierCrossed) return "synced";
const open = opens.get(nuri);
// Never opened here, or opened on the no-`doc_subscribe` path, which records nothing
// because there is no barrier to have missed.
if (open === undefined) return "unknown";
// Opened here, and no channel stands: the subscribe was refused, or a reset took it away,
// or `resetDocSubscriptions` dropped it from under us. Opened best-effort, sync UNconfirmed -
// never `"unknown"`, which is the reading that lets a zero-row read be called an absence.
if (channel.state === "none") return "timed-out";
return open.waiting ? "syncing" : "timed-out";
}
/**
@@ -294,69 +339,71 @@ export async function ensureRepoOpen(nuri: Nuri, holderKey?: string): Promise<vo
export async function openRepoUnguarded(nuri: Nuri): Promise<void> {
// Drop the registry if the session changed (in-page re-login → fresh verifier).
await syncSession();
// IN FLIGHT first, so a concurrent caller joins the open and waits on the same barrier
// rather than stepping past it: `held` is set the moment the subscription is taken out,
// which is before the `State` that makes it authoritative.
const pending = inFlight.get(nuri);
if (pending) return pending;
if (held.has(nuri)) return;
const standing = opens.get(nuri);
// A caller arriving while the bounded wait is still running JOINS it, and waits on the same
// barrier rather than stepping past it.
if (standing?.waiting === true) return standing.done;
// A caller arriving when this module's own listener is still in place AND the channel behind
// it has ANSWERED has nothing to do. Both halves, and each catches a defect the other
// cannot: without the listener half, a document whose subscription a reset took away would
// never be re-held for the new session; without the channel half - which is the whole of
// this task - a document whose `doc_subscribe` has not answered, or whose fan-out was torn
// down by a `configure()`, reads as open for ever and nothing re-subscribes it.
if (standing !== undefined && standing.release !== null && docChannel(nuri).state === "open") return;
// 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.
// does not regress. No record is kept (getSyncState → "unknown"): the fake
// path has no barrier semantics, and claiming a barrier here would be a lie.
const ng = getConfig().ng as { doc_subscribe?: unknown };
if (typeof ng.doc_subscribe !== "function") {
// Nothing to open and nothing to record: there is no subscription to hold, so `held`
// must stay empty for this document or it would claim one. No sync state either -
// `"unknown"` is exactly right here, and it is the one place that MEANS it.
// Nothing to open and nothing to record: there is no subscription to hold, so this
// document must stay out of `opens` or it would claim one - and `"unknown"` is exactly
// right here, the one place that MEANS "no barrier was ever expected".
return;
}
const myEpoch = registryEpoch;
// 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();
// Published BEFORE the subscribe, so a concurrent caller joins this wait instead of opening
// a second time - and carrying nothing but the wait, because at this instant nothing is
// true yet. What the previous registry put here was a claim of openness.
const mine: BootstrapOpen = { done: Promise.resolve(), release: null, settle: null, waiting: true };
opens.set(nuri, mine);
const p = (async () => {
mine.done = (async () => {
await new Promise<void>((resolve) => {
let settled = false;
// Handed to `resetOpenedRepos`, so an open this session abandons can be ended
// rather than left hanging on a subscription that no longer exists.
settleInFlight.set(nuri, () => {
if (settled) return;
settled = true;
resolve();
});
// Resolve on the first `State` (the sync BARRIER), marking the nuri "synced".
const onState = (): void => {
// The first `State` IS the barrier, WHENEVER it arrives. Recorded before the
// already-settled check on purpose: the bounded fallback below only means "this
// open stopped waiting", never "the barrier will not be crossed" - the
// subscription is still held, so a `State` can still land afterwards. It used to
// be dropped, which left `timed-out` standing for ever on a repo that had since
// synced, and a reader consulting that state (`surface/use-shape-query.ts`) could
// never learn better. `timed-out` -> `synced` is the only transition this adds,
// and it is the one that goes from less authority to more.
syncState.set(nuri, "synced");
if (settled) return;
settled = true;
logStage("BARRIER " + shortNuri(nuri) + " synced (" + (Date.now() - barrierStartedAt) + "ms)");
// Stop waiting, once. Handed to `resetOpenedRepos` as well, so an open this session
// abandons can be ended rather than left hanging on a subscription that no longer
// exists.
const stopWaiting = (): void => {
if (!mine.waiting) return;
mine.waiting = false;
mine.settle = null;
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.
mine.settle = stopWaiting;
// The first `State` IS the barrier, WHENEVER it arrives - and it is the FAN-OUT that
// records it, not this module, so a `State` landing long after this wait gave up still
// counts and still turns `getSyncState` to `"synced"`. That used to need a write placed
// carefully above the already-settled check; there is nothing to place now.
const onState = (): void => {
if (mine.waiting) {
logStage("BARRIER " + shortNuri(nuri) + " synced (" + (Date.now() - barrierStartedAt) + "ms)");
}
stopWaiting();
};
// Bounded fallback: proceed WITHOUT a `State`. A genuinely-absent doc reads 0 rows
// anyway (same as before, never a hang); giving up the WAIT is not crossing the
// barrier, and `getSyncState` says so because the fan-out has no `State` to show.
const onTimeout = (): void => {
if (settled) return;
settled = true;
syncState.set(nuri, "timed-out");
logStage("BARRIER " + shortNuri(nuri) + " timed-out (" + (Date.now() - barrierStartedAt) + "ms)");
resolve();
if (mine.waiting) {
logStage("BARRIER " + shortNuri(nuri) + " timed-out (" + (Date.now() - barrierStartedAt) + "ms)");
}
stopWaiting();
};
// 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
@@ -366,43 +413,53 @@ export async function openRepoUnguarded(nuri: Nuri): Promise<void> {
// rule 2 above; `ensurePhysicalRepoOpen` is the machinery's door and is not
// subject to the boundary at all (see physical.ts).
// A failed establish is REPORTED, and `subscribe.ts` offers this channel precisely
// so that a caller can retry. Ignoring it is what made a refused subscribe look like
// an open repo: the fallback below fired, the open was recorded, and every later
// attempt short-circuited on a subscription that had never existed.
// so that a caller can retry. Let the listener go, so the next caller opens a NEW
// subscription rather than joining a fan-out that is not going to push. There is no
// barrier state to correct: none was ever written here, and the fan-out has no `State`
// to show for a call that failed.
//
// NOT a substitute for the channel question above, and that is the lesson of this
// task: `subscribe.ts` reports a failure only for the attempt it is still WAITING ON,
// so a rejection arriving after `resetDocSubscriptions` released the fan-out is logged
// and nothing else. When the record was a claim of openness, that claim then stood for
// the whole session and no retry could ever get past it.
const onSetupFailed = (error: unknown): void => {
if (myEpoch !== registryEpoch) return;
if (opens.get(nuri) !== mine) return;
console.error(accessLogPrefix() + " ensureRepoOpen: subscribe failed for " + shortNuri(nuri) + ":", error);
// Record NOTHING, so the next caller subscribes again - and say plainly that no
// barrier was crossed, rather than leaving a state that reads as authority.
const taken = held.get(nuri);
if (taken) {
held.delete(nuri);
try {
taken();
} catch {
/* ignore */
}
const listening = mine.release;
mine.release = null;
try {
listening?.();
} catch {
/* ignore */
}
syncState.set(nuri, "timed-out");
settleInFlight.get(nuri)?.();
stopWaiting();
};
const unsub = subscribeDocUnguarded(
// The bootstrap subscription is kept ALIVE for the session - holding it open is the
// whole point. It is a LISTENER on the document's one real subscription, which is why
// it says nothing about whether that subscription exists.
mine.release = subscribeDocUnguarded(
nuri,
(_r, type) => {
if (type === "State") onState();
},
onSetupFailed,
);
held.set(nuri, unsub);
// A record this open REPLACED still holds a listener, and its wait is over - so nothing
// will ever release it. Done after the new listener has joined, so the fan-out keeps its
// entry and its establish, and this does not turn into a second `doc_subscribe` on a
// branch that already has one (which upstream would EVICT the first).
if (standing !== undefined && standing.release !== null) {
const superseded = standing.release;
standing.release = null;
try {
superseded();
} catch {
/* ignore */
}
}
setTimeout(onTimeout, OPEN_TIMEOUT_MS);
});
// Only if this open still belongs to the registry it started in. After a reset these
// maps are somebody else's, and writing into them is how a new session came to
// short-circuit on the previous one's record.
if (myEpoch !== registryEpoch) return;
inFlight.delete(nuri);
settleInFlight.delete(nuri);
})();
inFlight.set(nuri, p);
return p;
return mine.done;
}