fix: un second doc_subscribe tuait le premier, et les inbox n'étaient lues qu'à la connexion
This commit is contained in:
Sylvain Duchesne
@@ -0,0 +1,346 @@
|
||||
/**
|
||||
* While an identity is connected, its inboxes are WATCHED and what arrives is APPLIED.
|
||||
*
|
||||
* ── The regime this restores, and the one it replaces ─────────────────────
|
||||
* Upstream, applying an inbox is not something that happens at connection — it is what a
|
||||
* session DOES. A sealed message reaches the recipient's own verifier as it arrives
|
||||
* (`LocalBrokerMessage::Inbox` → `session.verifier.inbox(msg, from_queue)`,
|
||||
* `sdk/rust/src/local_broker.rs`), which unseals it and applies it inline
|
||||
* (`engine/verifier/src/inbox_processor.rs`); `from_queue` distinguishes the backlog handed
|
||||
* over at connection from the messages that follow, and BOTH go through the same door. So
|
||||
* the backlog is the special case, and the continuity is the rule.
|
||||
*
|
||||
* This package had emulated only the backlog. `inbox.processInbox` was called from exactly
|
||||
* one place — `connect.connectedUser`, at connection — and nothing anywhere applied a
|
||||
* deposit after that. A share deposited into an inbox whose owner was sitting connected in
|
||||
* front of it converged only when that person reloaded the page, and the deposit's own
|
||||
* session covered the gap by usurping the owner's identity on a twenty-second timer
|
||||
* (`emulated-verifier/inbox-processor.ts`) — which does nothing at all if the depositor
|
||||
* closes their tab, and tells the connected owner nothing either way.
|
||||
*
|
||||
* So: for as long as an identity is connected, every inbox it may read is subscribed to,
|
||||
* and every push over one of them applies what is in it.
|
||||
*
|
||||
* ── Watching is not applying ──────────────────────────────────────────────
|
||||
* `inbox.watch` looks like this and is not: it NOTIFIES an application that made a call and
|
||||
* named one inbox, and it applies nothing. This applies, on the whole set, without anybody
|
||||
* asking — because processing an inbox is the library's job and not the app's, which is the
|
||||
* same ruling `connect.ts` opens with.
|
||||
*
|
||||
* ── Subscription, never a poll ────────────────────────────────────────────
|
||||
* Every half is push-driven. The inboxes themselves are subscribed to individually. WHICH
|
||||
* inboxes exist is itself a subscription, on two channels that do not overlap:
|
||||
*
|
||||
* - the REGISTER — an inbox opened mid-session appends an `AddInboxCap` record to the
|
||||
* User branch of the private store (`branch-registers.openDocumentInbox`), so a push on
|
||||
* that document re-enumerates `myInboxes()`;
|
||||
* - WHAT THIS IDENTITY HOLDS (`CapRegistry.onChange`) — the same signal `watchShape`
|
||||
* re-reads on. It answers the case the register cannot: an identity that connects
|
||||
* before it has an account at all. Provisioning is lazy here, so a person's FIRST visit
|
||||
* connects with no stores, no private store to follow and no inbox to watch; the
|
||||
* account appears later, the moment the application creates anything, and filing its
|
||||
* caps is what says so. Without this channel a brand-new user was watched from their
|
||||
* SECOND visit onwards — which is exactly the person most likely to be sent something.
|
||||
*
|
||||
* So a document whose inbox this identity opened five minutes into the session is watched
|
||||
* like the rest, and so is the very first inbox of a person who had none when they arrived.
|
||||
*
|
||||
* ── It belongs to ONE holder, and stops when that holder does ─────────────
|
||||
* Everything below resolves the CURRENT holder when it runs — `myInboxes`, `processInbox`
|
||||
* and everything under them ask `getCurrentUser()` at the moment they execute. So an
|
||||
* observation started for Alice must not still be firing when Bob is connected: it would
|
||||
* read Alice's registers under Bob, or file Alice's capabilities into Bob's ring. The
|
||||
* observation is therefore stamped with its holder, every step re-checks it, and switching
|
||||
* identity (or disconnecting) tears it down — the same rule, and the same reason,
|
||||
* `connect.connectedUser` carries `stillConnected()` for.
|
||||
*/
|
||||
|
||||
import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap";
|
||||
import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log";
|
||||
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 type { Nuri, PrincipalId } from "../model/types";
|
||||
|
||||
/**
|
||||
* One identity's live observation. Held whole rather than as loose module variables so
|
||||
* that a step started under it can ask "am I still the current one" by IDENTITY of the
|
||||
* object, not by comparing a name that a re-connection may have restored in between.
|
||||
*/
|
||||
interface Observation {
|
||||
/** WHO this observation belongs to. Every step re-checks it before acting. */
|
||||
holder: PrincipalId;
|
||||
/** One subscription per inbox observed, keyed by it. */
|
||||
inboxes: Map<Nuri, Unsubscribe>;
|
||||
/** The subscription on the register that says WHICH inboxes exist. */
|
||||
register: Unsubscribe | null;
|
||||
/** Unsubscribe from the held-caps change signal — the second "which inboxes" channel. */
|
||||
holdings: Unsubscribe | null;
|
||||
/** 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. */
|
||||
enumerateAgain: boolean;
|
||||
/** Enumerations and applications in flight — what {@link observationSettled} waits on. */
|
||||
pending: Set<Promise<void>>;
|
||||
}
|
||||
|
||||
let observation: Observation | null = null;
|
||||
|
||||
/**
|
||||
* Where a failure to WATCH goes — distinct from a failure to APPLY, which
|
||||
* `inbox-drain.ts` reports.
|
||||
*
|
||||
* Not a rejection: nothing awaits this work, exactly as at connection. And not silence, for
|
||||
* the same reason as everywhere else in this package — an observation that cannot be
|
||||
* established leaves deposits unapplied for the whole session, and the only thing a person
|
||||
* would notice is that a share never arrives.
|
||||
*/
|
||||
function reportUnobserved(what: string, error: unknown): void {
|
||||
console.error(
|
||||
accessLogPrefix() + " connected, but " + what + " — deposits made from now on may not be " +
|
||||
"applied until the next connection:",
|
||||
error,
|
||||
);
|
||||
}
|
||||
|
||||
/** Is `obs` still the live observation, for the identity it belongs to? */
|
||||
function current(obs: Observation): boolean {
|
||||
return observation === obs && getCurrentUser() === obs.holder;
|
||||
}
|
||||
|
||||
/** Track a piece of in-flight work so {@link observationSettled} can wait for it. */
|
||||
function track(obs: Observation, work: Promise<void>): void {
|
||||
obs.pending.add(work);
|
||||
void work.finally(() => obs.pending.delete(work));
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply whatever is in `inbox`, for the holder this observation belongs to.
|
||||
*
|
||||
* Through the shared per-inbox queue, so this cannot interleave with a drain the deferred
|
||||
* timer started on the same inbox (nor with another push of its own): two passes over one
|
||||
* queue apply the same records twice. A failure is reported there and stops nothing — the
|
||||
* next push over this inbox tries again, and the other inboxes were never involved.
|
||||
*/
|
||||
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.
|
||||
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`.
|
||||
if (!current(obs)) return;
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* (Re-)enumerate the inboxes this identity may read and subscribe to the ones not yet
|
||||
* watched. Idempotent: an inbox already observed is left alone, so a re-enumeration costs
|
||||
* one read and nothing else.
|
||||
*
|
||||
* Never removes: the list only grows within a session (an `AddInboxCap` record is durable),
|
||||
* and dropping a subscription on a guess would silently stop applying an inbox.
|
||||
*/
|
||||
async function watchTheInboxes(obs: Observation): Promise<void> {
|
||||
if (!current(obs)) return;
|
||||
let inboxes: Nuri[];
|
||||
try {
|
||||
inboxes = await myInboxes();
|
||||
} catch (error) {
|
||||
// Not knowing WHICH inboxes exist is the whole observation failing, not one queue.
|
||||
reportUnobserved("the inboxes to watch could not be listed", error);
|
||||
return;
|
||||
}
|
||||
if (!current(obs)) return;
|
||||
for (const inbox of inboxes) {
|
||||
if (obs.inboxes.has(inbox)) continue;
|
||||
try {
|
||||
// The push is the signal; the drain is the work. The FIRST push is the initial
|
||||
// `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.
|
||||
obs.inboxes.set(
|
||||
inbox,
|
||||
subscribeDoc(inbox, () => track(obs, applyWhatArrived(obs, inbox))),
|
||||
);
|
||||
logStage("OBSERVING " + shortNuri(inbox) + " for " + obs.holder);
|
||||
} catch (error) {
|
||||
reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* the private store, and a same-session write pushes to its own subscribers.
|
||||
*
|
||||
* Idempotent, and RETRIED rather than given up on: an identity that connects before it has
|
||||
* an account has no private store to follow yet, and acquires one the first time the
|
||||
* application creates anything. Re-entered from the holdings signal, it catches up then.
|
||||
*/
|
||||
async function watchTheRegister(obs: Observation): Promise<void> {
|
||||
if (!current(obs) || obs.register !== null) return;
|
||||
let store: Nuri | undefined;
|
||||
try {
|
||||
// `lookupAccount`, not `resolveAccount`: the tolerant form answers `null` for a read
|
||||
// that FAILED exactly as for one that found nothing, and the two could not mean more
|
||||
// different things here — "this identity has no account yet" is a state to wait
|
||||
// quietly through, "the broker did not answer" is a failure to report. The same
|
||||
// distinction `connect.connectedUser` and `inbox.share` are built on.
|
||||
store = (await lookupAccount(obs.holder))?.docPrivate;
|
||||
} catch (error) {
|
||||
reportUnobserved("the register of your inboxes could not be reached", error);
|
||||
return;
|
||||
}
|
||||
// No account yet — nothing to follow, and asking for one would PROVISION it, which
|
||||
// connecting deliberately does not do (`connect.connectedUser`). The holdings signal
|
||||
// brings us back here the moment there is something to follow.
|
||||
if (!store || !current(obs) || obs.register !== null) return;
|
||||
try {
|
||||
obs.register = subscribeDoc(store, () => track(obs, enumerate(obs)));
|
||||
} catch (error) {
|
||||
reportUnobserved("an inbox opened later in this session will not be watched", error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The whole "which inboxes am I watching" cycle: catch up on the register, then on the
|
||||
* inboxes. Both halves are idempotent, so this is what every trigger runs.
|
||||
*
|
||||
* A trigger that arrives while a cycle is running does not start a second one — the cycle
|
||||
* FILES caps for the inboxes it lists (`myInboxes` → `fileOwnInbox`), which is one of the
|
||||
* signals that runs it, so overlapping runs would pile up on their own effects. It is
|
||||
* REMEMBERED instead of dropped: the running cycle repeats once more at the end, because a
|
||||
* signal that arrived mid-cycle may be about something that cycle had already read past.
|
||||
*
|
||||
* That terminates rather than ping-ponging: filing a cap this identity already holds fires
|
||||
* nothing (`caps.file` notifies only on a genuinely new one), so the repeat that finds
|
||||
* nothing new sets no flag and the loop ends.
|
||||
*/
|
||||
async function enumerate(obs: Observation): Promise<void> {
|
||||
if (!current(obs)) return;
|
||||
if (obs.enumerating) {
|
||||
obs.enumerateAgain = true;
|
||||
return;
|
||||
}
|
||||
obs.enumerating = true;
|
||||
try {
|
||||
do {
|
||||
obs.enumerateAgain = false;
|
||||
await watchTheRegister(obs);
|
||||
await watchTheInboxes(obs);
|
||||
} while (obs.enumerateAgain && current(obs));
|
||||
} finally {
|
||||
obs.enumerating = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start observing the connected identity's inboxes. Called at the end of the connection
|
||||
* work (`connect.connectedUser`), and idempotent: calling it again for the identity already
|
||||
* being observed changes nothing.
|
||||
*
|
||||
* ── It resolves when the WATCHING is in place, not when the applying is done ──
|
||||
* Connecting publishes one guarantee — *the work it starts, it finishes* — and "your
|
||||
* inboxes are now being watched" is part of that work, so this is awaited rather than
|
||||
* fired. What it does NOT wait for is the applying: subscribing pushes an initial `State`,
|
||||
* which drains each inbox again, and making a person's sign-in wait on a second pass over
|
||||
* a backlog `connectedUser` has just applied would buy nothing.
|
||||
*
|
||||
* It never rejects. Failing to WATCH is reported where failures go and denies nobody their
|
||||
* session — the same rule the drain loop above it follows, and for the same reason: what
|
||||
* has not been applied stays in its queue for the next attempt.
|
||||
*/
|
||||
// @provenance startObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream a connected verifier is handed each inbox message as it arrives and applies it; here the arrival signal is a document push, and the set of inboxes is the User branch's `AddInboxCap` records
|
||||
export async function startObservingInboxes(): Promise<void> {
|
||||
const holder = getCurrentUser();
|
||||
if (holder === null) return;
|
||||
if (observation !== null && observation.holder === holder) return;
|
||||
stopObservingInboxes();
|
||||
const obs: Observation = {
|
||||
holder,
|
||||
inboxes: new Map(),
|
||||
register: null,
|
||||
holdings: null,
|
||||
enumerating: false,
|
||||
enumerateAgain: false,
|
||||
pending: new Set(),
|
||||
};
|
||||
observation = obs;
|
||||
// The second "which inboxes exist" channel — see the module header. It is what makes a
|
||||
// person's FIRST session watched at all: they connect holding nothing, and their account
|
||||
// (with its inboxes) appears the moment the application creates something.
|
||||
obs.holdings = getCaps().onChange(() => track(obs, enumerate(obs)));
|
||||
const setup = enumerate(obs);
|
||||
track(obs, setup);
|
||||
await setup;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop observing — on an identity change, on disconnection, and when the library is
|
||||
* un-configured.
|
||||
*
|
||||
* Work already inside the drain queue is not cancelled (nothing can un-issue a write
|
||||
* already sent); it is neutralised instead, because every step re-checks that its
|
||||
* observation is still the current one and returns rather than filing for the wrong holder.
|
||||
*/
|
||||
// @provenance stopObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:Verifier — a verifier's inbox processing lives and dies with its session; switching identity is another session, so nothing of the previous one keeps applying
|
||||
export function stopObservingInboxes(): void {
|
||||
const obs = observation;
|
||||
if (obs === null) return;
|
||||
observation = null;
|
||||
for (const unsub of obs.inboxes.values()) {
|
||||
try {
|
||||
unsub();
|
||||
} catch (error) {
|
||||
console.error(accessLogPrefix() + " stopping an inbox observation failed:", error);
|
||||
}
|
||||
}
|
||||
obs.inboxes.clear();
|
||||
for (const unsub of [obs.register, obs.holdings]) {
|
||||
if (!unsub) continue;
|
||||
try {
|
||||
unsub();
|
||||
} catch (error) {
|
||||
console.error(accessLogPrefix() + " stopping an inbox observation failed:", error);
|
||||
}
|
||||
}
|
||||
obs.register = null;
|
||||
obs.holdings = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait until the observation has nothing in flight — the enumerations it is running and the
|
||||
* drains they started.
|
||||
*
|
||||
* A seam for a suite that has to observe convergence without waiting for it, in the same
|
||||
* spirit as `runScheduledInboxProcessingNow`: it names no inbox and starts no work, so it
|
||||
* can only wait for what a push has already caused. Never exported from the package.
|
||||
*/
|
||||
// @provenance observationSettled kind=invention level=none ref=none — a test seam over the observation; upstream nothing exposes "has the verifier finished applying"
|
||||
export async function observationSettled(): Promise<void> {
|
||||
for (;;) {
|
||||
const obs = observation;
|
||||
const inFlight = obs ? [...obs.pending] : [];
|
||||
if (inFlight.length === 0) break;
|
||||
await Promise.all(inFlight);
|
||||
}
|
||||
await drainsSettled();
|
||||
}
|
||||
Reference in New Issue
Block a user