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

fix: un abonnement en échec n'empoisonne plus le document, et un dépôt ne change plus de destinataire

Browse Source
This commit is contained in:
Sylvain Duchesne
2026-08-17 11:47:21 +02:00
parent 520c8c59a8
commit 33212a8b00
7 changed files with 493 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/polyfill/src/emulated-verifier/connect.ts
+43 -22
View File
@@ -18,7 +18,9 @@
* new Link durably and puts it among what the user holds.
* 3. **Keep applying** — start watching those same inboxes, for as long as this identity
* stays connected (`emulated-verifier/inbox-observer.ts`). Step 2 is the backlog;
* this is the regime.
* this is the regime. It runs whatever became of the first two, including a step 1 that
* REJECTED: what is watched is *being connected*, and `setCurrentUser` has already made
* that true by the time any of this runs.
*
* Restoring first means a reconnecting user can read its documents immediately, without
* waiting on the inbox round-trip; watching last means the backlog is applied before the
@@ -164,7 +166,7 @@ export async function connectedUser(): Promise<void> {
// Captured with the identity, handed back at filing time — see `caps.holderKey`.
const holderKey = getCaps().holderKey();
const run = (async (): Promise<void> => {
const restoreAndDrain = async (): Promise<void> => {
// Connecting must not PROVISION. `ensureAccount` would create the user on
// first sight, so connecting an identity that does not exist yet would
// silently mint its stores and their caps — arming the whole emulation as a
@@ -186,8 +188,7 @@ export async function connectedUser(): Promise<void> {
// answers `[]` for an identity with no account) and is what makes that first session
// behave like every other one — the alternative was a brand-new user watched from
// their SECOND visit onwards, which is precisely the person most likely to be sent
// something.
if (stillConnected()) await startObservingInboxes();
// something. (Started below, for every outcome of this function alike.)
return;
}
if (!stillConnected()) return;
@@ -226,24 +227,44 @@ export async function connectedUser(): Promise<void> {
reportUndrained(inbox, error);
}
}
// 3. …and from here on, KEEP applying. The backlog above is the special case, not the
// rule: upstream a session is handed each inbox message as it arrives and applies it
// inline, and only the messages waiting at connection are the "queue"
// (`from_queue`). Draining once and stopping made a deposit wait for the recipient
// to reload the page. See `emulated-verifier/inbox-observer.ts`.
//
// After the drain, not before: connecting owes the backlog first, and the
// observation subscribes to the same inboxes this loop just read.
//
// Unconditional on the loop's outcome, deliberately. An inbox that could not be
// drained is reported and denies nobody their session, and it must not deny them the
// observation of the OTHER inboxes either — nor of itself, since the next push is a
// fresh attempt at exactly the deposit that failed.
// Awaited, and it never rejects: what connecting starts, connecting finishes, so a
// caller that got its promise back knows the watching is in place — not merely
// requested. (It does not wait for the applying that watching then triggers.)
if (!stillConnected()) return;
await startObservingInboxes();
};
const run = (async (): Promise<void> => {
try {
await restoreAndDrain();
} finally {
// 3. …and from here on, KEEP applying. The backlog above is the special case, not the
// rule: upstream a session is handed each inbox message as it arrives and applies it
// inline, and only the messages waiting at connection are the "queue"
// (`from_queue`). Draining once and stopping made a deposit wait for the recipient
// to reload the page. See `emulated-verifier/inbox-observer.ts`.
//
// After the restore and the drain, not before: connecting owes the backlog first,
// and the observation subscribes to the same inboxes that loop just read.
//
// ── Unconditional on how they WENT, and that is the whole point ──
// An inbox that could not be drained is reported and denies nobody their session,
// and it must not deny them the observation of the OTHER inboxes either — nor of
// itself, since the next push is a fresh attempt at exactly the deposit that failed.
//
// The same holds one step up, and until 2026-08-17 it did not: a restore that
// REJECTED skipped this line, and the identity was left CONNECTED — `setCurrentUser`
// is synchronous and had already taken effect — with nothing watching its inboxes
// for the rest of the session. `startObservingInboxes` has no other caller, so one
// broker hiccup at sign-in cost that person every deposit made from then on, in
// silence, long after the broker had recovered.
//
// It does not blur the rule this function is built on. The rule is about what the
// CALLER is told — failing to reach the queues rejects, failing to apply one is
// reported — and rejecting is exactly what still happens: the error raised above
// propagates through this `finally` untouched. What changes is that being connected
// now means being watched, whatever the connection made of its own work.
//
// Awaited, and it never rejects: what connecting starts, connecting finishes, so a
// caller that got its promise back knows the watching is in place — not merely
// requested. (It does not wait for the applying that watching then triggers.)
if (stillConnected()) await startObservingInboxes();
}
})();
inFlight.set(holder, run);
packages/polyfill/src/emulated-verifier/inbox-observer.ts
+72 -13
View File
@@ -61,7 +61,7 @@ import { myInboxes } from "./branch-registers";
import { drainInboxSerially, drainsSettled } from "./inbox-drain";
import { lookupAccount } from "../shared-wallet/account-registry";
import { processInbox } from "../surface/inbox";
import { subscribeDoc, type Unsubscribe } from "../surface/subscribe";
import { subscribeDoc, subscribeDocReportingSetupFailure, type Unsubscribe } from "../surface/subscribe";
import type { Nuri, PrincipalId } from "../model/types";
/**
@@ -78,6 +78,12 @@ interface Observation {
register: Unsubscribe | null;
/** Unsubscribe from the held-caps change signal — the second "which inboxes" channel. */
holdings: Unsubscribe | null;
/**
* The inboxes whose subscription has already been re-attempted once after failing to be
* OPENED. What bounds the repair below to one extra try per inbox, so a document the broker
* genuinely cannot serve costs two calls rather than a loop.
*/
reattempted: Set<Nuri>;
/** True while a re-enumeration is running, so its own effects do not restart it. */
enumerating: boolean;
/** A trigger that arrived mid-cycle: the running one repeats once rather than lose it. */
@@ -128,21 +134,30 @@ async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise<void> {
if (!current(obs)) return;
await drainInboxSerially(inbox, async () => {
// Re-checked INSIDE the queue: this run may have waited behind another one, and the
// identity can have moved while it waited. `processInbox` resolves the holder at each
// of its steps, so running it for the wrong one is not a near-miss — it reads someone
// else's registers and files into someone else's ring.
// identity can have moved while it waited. Running it for the wrong holder is not a
// near-miss — it reads someone else's registers and files into someone else's ring.
//
// This check is NECESSARY and it is not SUFFICIENT, and until 2026-08-17 this comment
// claimed it was. It cannot be: the identity can move after it passes, while
// `processInbox` is mid-read. The claim was that `processInbox` "resolves the holder at
// each step, so the new holder is refused an inbox that is not theirs somewhere in the
// middle" — but its ownership guard runs ONCE, at entry, and a switch after that reached
// the filing with nobody left to refuse it. What it filed was the previous holder's cap,
// into the new holder's ring, durably. The guard that makes this run safe is therefore
// the one INSIDE `processInbox`, which captures the holder its guard authorised; this one
// only spares the work when the switch is already visible.
if (!current(obs)) return;
try {
await processInbox(inbox);
} catch (error) {
// An identity that moved MID-DRAIN throws here too, and it is the ordinary case rather
// than an exotic one: `processInbox` resolves the holder at each step, so the new
// holder is refused an inbox that is not theirs somewhere in the middle. That is
// ABANDONING — which this package has always called "not a failure" — and reporting it
// would put a broker-looking error in the log every time a page switches user.
// Nothing is lost: an inbox is not consumed by being abandoned, so the deposit is
// still there for its owner's next connection. Same rule, same words, as the drain
// loop in `connect.connectedUser`.
// An identity that moved MID-DRAIN can throw here too — the switch may land before
// `processInbox`'s own ownership guard, which then refuses the new holder an inbox that
// is not theirs. (Landing after it, the run ABANDONS quietly instead and returns; both
// leave the deposit where it is.) Abandoning is what this package has always called
// "not a failure", and reporting it would put a broker-looking error in the log every
// time a page switches user. Nothing is lost: an inbox is not consumed by being
// abandoned, so the deposit is still there for its owner's next connection. Same rule,
// same words, as the drain loop in `connect.connectedUser`.
if (!current(obs)) return;
throw error;
}
@@ -175,9 +190,19 @@ async function watchTheInboxes(obs: Observation): Promise<void> {
// `State`, which drains an inbox connection has usually just drained — idempotent,
// and the alternative (skip the first) would lose a deposit that landed in the gap
// between the two.
//
// The failure channel is not decoration: `subscribeDoc` returns SYNCHRONOUSLY and its
// one real `doc_subscribe` is opened afterwards, so a rejection there reaches nobody.
// Without it, this map held an entry for an inbox that was never watched, the loop
// above skipped it at every later enumeration, and the only observable difference from
// a healthy session was that shares stopped arriving.
obs.inboxes.set(
inbox,
subscribeDoc(inbox, () => track(obs, applyWhatArrived(obs, inbox))),
subscribeDocReportingSetupFailure(
inbox,
() => track(obs, applyWhatArrived(obs, inbox)),
(error) => watchFailed(obs, inbox, error),
),
);
logStage("OBSERVING " + shortNuri(inbox) + " for " + obs.holder);
} catch (error) {
@@ -186,6 +211,39 @@ async function watchTheInboxes(obs: Observation): Promise<void> {
}
}
/**
* The subscription on `inbox` could not be OPENED — report it, forget it, and try once more.
*
* **Forget it**, because the entry this observation holds is the whole record of "already
* watched": leaving a dead one in place is how one rejection at connection turned into a
* session-long silence. Removed, the next enumeration subscribes again as if it had never
* been attempted.
*
* **Try once more**, because the usual cause is a repo the verifier has not synced yet, and
* the whole cost of finding out is one call. Exactly one extra attempt per inbox, tracked in
* {@link Observation.reattempted}: a broker that genuinely cannot serve this document must
* not be asked again and again by a loop that has no reason to stop. If the second attempt
* fails too, the failure has been reported twice and the inbox waits for the next
* enumeration signal — or for the next connection, where an unapplied deposit has always
* waited.
*/
function watchFailed(obs: Observation, inbox: Nuri, error: unknown): void {
if (!current(obs)) return;
const stale = obs.inboxes.get(inbox);
obs.inboxes.delete(inbox);
if (stale) {
try {
stale();
} catch (thrown) {
console.error(accessLogPrefix() + " releasing a failed inbox observation failed:", thrown);
}
}
reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error);
if (obs.reattempted.has(inbox)) return;
obs.reattempted.add(inbox);
track(obs, enumerate(obs));
}
/**
* Follow the register that says which inboxes exist, so one opened MID-SESSION is picked
* up. See the module header: `openDocumentInbox` appends its record to the User branch of
@@ -279,6 +337,7 @@ export async function startObservingInboxes(): Promise<void> {
inboxes: new Map(),
register: null,
holdings: null,
reattempted: new Set(),
enumerating: false,
enumerateAgain: false,
pending: new Set(),
packages/polyfill/src/surface/inbox.ts
+32 -1
View File
@@ -669,16 +669,47 @@ export async function readSyncedForDocument(docLike: NuriLike): Promise<Deposit[
* Idempotent: `addLink` ignores a Link it already holds, so processing twice (a
* second tab, a reconnect) costs nothing. Returns the consumer deposits, exactly as
* {@link read} does — Links are never surfaced.
*
* ── The holder is captured, not re-resolved after the read ────────────────
* The ownership guard runs at entry (inside {@link readSynced}) and the FILING happens a
* broker round-trip later, so re-asking `getCurrentUser()` at that point asks a different
* question: not "whose inbox is this" but "who is holding the page NOW". A page that
* switched identity in that window filed CAROL's cap, deposited for ALICE, durably into
* BOB's ring — `documentsGivenTo("bob")` returns Carol's document, Alice gets nothing, and
* Bob's next sign-in restores it as a capability he holds. Not a near-miss: one user is
* given another's document and the real recipient is left with none.
*
* So the holder the guard authorised is the one the filing is checked against, and a switch
* ABANDONS — the same answer {@link read} gives for its own half of this window, and the
* same one `connect.connectedUser` gives for a drain it started. Nothing is lost: an inbox
* is not consumed by being read, so the deposit is still there for its owner's next
* connection or its next push. Filing it FOR the absent owner instead would be a write on
* somebody else's store from a published call — which is the deferred processor's
* deliberate, unpublished privilege (`emulated-verifier/inbox-processor.ts`), not this one's.
*/
// @provenance inbox.processInbox kind=aligned level=1 ref=engine/verifier/src/inbox_processor.rs:process_inbox — unseal-and-apply: Links are filed durably and never surfaced. Returning the remaining deposits is the divergent half it shares with `read`
export async function processInbox(targetInboxLike: NuriLike): Promise<Deposit[]> {
const targetInbox = toNuri(targetInboxLike, "inbox.processInbox");
// WHO this processing is for, captured before the read that authorises it — see above.
const holder = getCurrentUser();
const deposits = await readSynced(targetInbox);
// `readSynced` already put every Link in memory for this session; now make
// them durable. Reading the raw deposits again would mean re-parsing, so the caps
// are taken from what the read just observed.
for (const cap of capsSeenIn(targetInbox)) await addLink(cap);
const seen = capsSeenIn(targetInbox);
seenByInbox.delete(targetInbox);
for (const cap of seen) {
// Re-checked per cap, not once: `addLink` reads and writes, so the identity can move
// between two of them just as easily as during the read.
if (getCurrentUser() !== holder) {
logStage(
"ABANDONED " + shortNuri(targetInbox) + " — the identity changed while it was being " +
"processed; its deposits stay for their owner",
);
return deposits;
}
await addLink(cap, holder ?? undefined);
}
return deposits;
}
packages/polyfill/src/surface/subscribe.ts
+128 -11
View File
@@ -61,6 +61,19 @@
* it — without the replay, joining a document somebody else already opened would never
* fire, and `inbox.watch`'s "fires once immediately" would silently stop being true.
* The real subscription is torn down when the LAST listener leaves.
*
* ── What SHARING one subscription must not cost a caller ───────────────────
* Three things a per-caller `doc_subscribe` gave for free, and that the fan-out has to give
* back deliberately — each of them was lost when it was introduced on 2026-08-17:
*
* - a caller is ITSELF, not its handler. One record per CALL, so passing the same function
* twice is two subscriptions and the first unsubscribe does not silence the second;
* - a caller that has LEFT hears nothing more, including from the push during which it
* left — a handler may tear another one down, and the pushes are re-checked against the
* live set rather than a copy taken before the first handler ran;
* - a setup that FAILED is not the end of the document. The attempt is forgotten so the
* next joiner makes its own (upstream, each caller's own `doc_subscribe` retried), and
* it is reported rather than left as silence — see {@link reportSetupFailure}.
*/
import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap";
@@ -122,16 +135,51 @@ async function sessionId(): Promise<string | number> {
/** What a listener is handed on every push. */
type Listener = (r: DocChange, type: DocChangeType) => void;
/**
* ONE call to {@link subscribeDocUnguarded}, and what that caller asked for.
*
* Identified by this record and never by the `onChange` function: two callers may
* legitimately pass the SAME function — a module-level handler, a bound method, an arrow
* that closes over nothing — and they are two subscriptions with two independent lifetimes.
* Keyed by the function, the second call was swallowed by the set and the FIRST caller's
* unsubscribe silenced the second, which had never asked to leave.
*/
interface DocListener {
/** What this caller is handed on every push. */
onChange: Listener;
/**
* Told when the one real `doc_subscribe` behind this listener could not be OPENED.
*
* Upstream `doc_subscribe` is async and REJECTS on a setup failure, so its caller learns.
* This wrapper returns synchronously, so without this channel a caller cannot tell "this
* document is quiet" from "this document is not subscribed at all" — the failure-as-absence
* this package keeps closing. `null` for a caller that did not ask (the published
* {@link subscribeDoc}, whose contract has no such argument); the failure is logged either
* way.
*/
onSetupFailed: ((error: unknown) => void) | null;
}
/**
* The single real `doc_subscribe` behind every local listener on one document.
* See the module header for why there can only be one.
*/
interface DocFanOut {
/** Every local listener on this document. The last one to leave tears it down. */
listeners: Set<Listener>;
listeners: Set<DocListener>;
/** The platform's unsubscribe, once the async setup resolved. */
realUnsub: (() => void) | null;
/** True from the moment setup is kicked off — a later joiner must not kick off a second. */
/**
* True from the moment a setup is kicked off — a later joiner must not kick off a second,
* because a second `doc_subscribe` on a branch EVICTS the first (see the module header).
* It therefore stays true once the setup has SUCCEEDED: the attempt still stands.
*
* Cleared on the two ways the attempt stops standing: the last listener leaving
* ({@link releaseFanOut}), and a setup that FAILED ({@link reportSetupFailure}). Left true
* on failure it stopped meaning "one is already running" and started meaning "this NURI is
* finished" — no later joiner ever attempted it again, for the whole session, over one
* transient rejection.
*/
establishing: boolean;
/**
* The most recent `State` push, replayed to a listener that joins later.
@@ -143,14 +191,57 @@ interface DocFanOut {
const fanOuts = new Map<Nuri, DocFanOut>();
/** Hand one push to one listener, isolating a throwing handler from the others. */
function deliver(nuri: Nuri, listener: Listener, resp: DocChange, type: DocChangeType): void {
function deliver(nuri: Nuri, listener: DocListener, resp: DocChange, type: DocChangeType): void {
try {
listener(resp, type);
listener.onChange(resp, type);
} catch (error) {
console.error("[subscribe] onChange handler threw for", nuri, error);
}
}
/**
* Hand one push to every listener of `entry` — the fan-out itself.
*
* Over a COPY, because a handler may unsubscribe itself or another from inside the push; and
* re-checking each listener against the live set, because a copy alone only stops the walk
* from breaking — it still delivers to whoever left DURING it. That contradicted what
* {@link subscribeDoc} publishes ("no further `onChange` fires after unsubscribe"), on the
* one ordering an application cannot control: whether its handler runs before or after the
* one that tore it down.
*/
function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeType): void {
for (const listener of [...entry.listeners]) {
if (!entry.listeners.has(listener)) continue;
deliver(nuri, listener, resp, type);
}
}
/**
* The one real subscription could not be OPENED: forget the attempt, and say so.
*
* Two halves, and the shape of every "failure disguised as an absence" this package has
* closed. **Retryable** — `establishing` goes back to false, so the next `subscribeDoc` on
* this NURI attempts it again instead of joining a fan-out that will never push. Left true,
* one transient rejection made the document silent for every later joiner in the session.
* **Visible** — the log always, plus the callers who asked to be told, so a caller whose job
* depends on the subscription (the inbox observation) can report it and try again rather
* than sit on a dead entry.
*/
function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown): void {
console.error("[subscribe] doc_subscribe failed for", nuri, error);
entry.establishing = false;
if (fanOuts.get(nuri) !== entry) return;
for (const listener of [...entry.listeners]) {
const tell = listener.onSetupFailed;
if (tell === null || !entry.listeners.has(listener)) continue;
try {
tell(error);
} catch (thrown) {
console.error("[subscribe] onSetupFailed handler threw for", nuri, thrown);
}
}
}
/** Drop a fan-out: forget the remembered state and release the real subscription. */
function releaseFanOut(nuri: Nuri, entry: DocFanOut): void {
if (fanOuts.get(nuri) === entry) fanOuts.delete(nuri);
@@ -187,8 +278,7 @@ async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unk
const type = docChangeType(resp);
// Remember the barrier for whoever joins next; a later `State` replaces it.
if (type === "State") entry.lastState = { resp, type };
// A copy: a handler may unsubscribe itself (or another) from inside the push.
for (const listener of [...entry.listeners]) deliver(nuri, listener, resp, type);
fanOut(nuri, entry, resp, type);
})) as (() => void) | undefined;
if (fanOuts.get(nuri) !== entry || entry.listeners.size === 0) {
// Everyone left (or the fan-out was reset) before setup resolved — cancel now.
@@ -197,7 +287,7 @@ async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unk
}
entry.realUnsub = typeof unsub === "function" ? unsub : null;
} catch (error) {
console.error("[subscribe] doc_subscribe failed for", nuri, error);
reportSetupFailure(nuri, entry, error);
}
}
@@ -291,6 +381,30 @@ export function subscribeDoc(
return subscribeDocUnguarded(nuri, onChange);
}
/**
* {@link subscribeDoc}, for a caller that must be TOLD when the subscription could not be
* opened. Same guard, same fan-out; the only difference is that a setup failure reaches
* `onSetupFailed` instead of only the log.
*
* Internal, and deliberately not the published shape: upstream `doc_subscribe` is async and
* rejects, so a failure has a caller to reach; ours returns synchronously, and the published
* signature has no room for it (`.project/concepts/app-contract/polyfill-surface/`). An
* application uses a subscription as a change SIGNAL and has nothing to do with the answer,
* so it keeps the two-argument call. The inbox observation does have something to do with it
* — an inbox it believes it is watching and is not leaves every deposit unapplied for the
* session — so it asks.
*/
// @provenance subscribeDocReportingSetupFailure kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — restores to an internal caller what upstream's async `doc_subscribe` gives every caller: the setup failure. One delta remains, the synchronous unsubscribe
export function subscribeDocReportingSetupFailure(
nuriLike: NuriLike,
onChange: (r: DocChange, type: DocChangeType) => void,
onSetupFailed: (error: unknown) => void,
): Unsubscribe {
const nuri = toNuri(nuriLike, "subscribeDoc");
assertMayReach(nuri, "subscribeDoc");
return subscribeDocUnguarded(nuri, onChange, onSetupFailed);
}
/**
* The unguarded core. Exported for ONE importer — `shared-wallet/physical.ts`, which
* owns the machinery's entire privileged door — and for nobody else. It is not
@@ -300,6 +414,7 @@ export function subscribeDoc(
export function subscribeDocUnguarded(
nuri: Nuri,
onChange: (r: DocChange, type: DocChangeType) => void,
onSetupFailed?: (error: unknown) => void,
): Unsubscribe {
// Resolved here, synchronously, so calling this before `configure()` still throws at
// the call rather than inside a background task nobody awaits.
@@ -311,7 +426,9 @@ export function subscribeDocUnguarded(
fanOuts.set(nuri, entry);
}
const joined = entry;
joined.listeners.add(onChange);
// THIS call's subscription — see {@link DocListener} for why it is not the function.
const listener: DocListener = { onChange, onSetupFailed: onSetupFailed ?? null };
joined.listeners.add(listener);
// Joining a document somebody else already opened: hand this listener the `State` its
// own `doc_subscribe` would have pushed it. Asynchronously, like the real push, so a
@@ -319,8 +436,8 @@ export function subscribeDocUnguarded(
if (joined.lastState) {
const { resp, type } = joined.lastState;
queueMicrotask(() => {
if (fanOuts.get(nuri) === joined && joined.listeners.has(onChange)) {
deliver(nuri, onChange, resp, type);
if (fanOuts.get(nuri) === joined && joined.listeners.has(listener)) {
deliver(nuri, listener, resp, type);
}
});
}
@@ -334,7 +451,7 @@ export function subscribeDocUnguarded(
return () => {
if (stopped) return;
stopped = true;
joined.listeners.delete(onChange);
joined.listeners.delete(listener);
// The LAST one out releases the real subscription — while anybody is still
// listening, tearing it down would silence them (and re-opening it later is not
// free: a second `doc_subscribe` evicts whoever else has the branch by then).