fix: dire le pari comme un pari, et distinguer deux manques d'inbox que rien ne distinguait
This commit is contained in:
@@ -502,10 +502,29 @@ export async function myInboxes(): Promise<Nuri[]> {
|
||||
export interface InboxEnumeration {
|
||||
/** Every inbox this holder may read that COULD be listed — possibly not all of them. */
|
||||
inboxes: Nuri[];
|
||||
/** What stopped the rest from being listed, or `null` when nothing did. */
|
||||
incomplete: { error: unknown } | null;
|
||||
/**
|
||||
* What stopped the rest from being listed, or `null` when nothing did — and WHICH of the
|
||||
* two registers stopped it, because the two shortfalls are not the same condition.
|
||||
*
|
||||
* `"account-record"` means the user's OWN two inboxes are missing, which is where a share
|
||||
* addressed to a PERSON lands: the identity has nothing watched that a stranger can reach,
|
||||
* and the document half was never even attempted (the account record is what it reads
|
||||
* through). `"user-branch"` means those two are in hand and only the per-document inboxes
|
||||
* are missing. Told apart because a caller that reports one of them has to be able to say
|
||||
* that the OTHER one has now happened — `error` alone cannot, and the observation
|
||||
* de-duplicating on "already said something" swallowed exactly that.
|
||||
*/
|
||||
incomplete: { error: unknown; register: InboxRegister } | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* WHICH of the two registers behind {@link enumerateMyInboxes} could not be read.
|
||||
*
|
||||
* Named after the register, not after the failure, because that is what a reader has to go
|
||||
* and look at: the account record in the doc-shim, and the User branch of the private store.
|
||||
*/
|
||||
export type InboxRegister = "account-record" | "user-branch";
|
||||
|
||||
/**
|
||||
* {@link myInboxes}, for the caller that can use a PARTIAL answer — the live observation.
|
||||
*
|
||||
@@ -537,7 +556,7 @@ export async function enumerateMyInboxes(): Promise<InboxEnumeration> {
|
||||
} catch (error) {
|
||||
// The account record is what the other half reads THROUGH (`readInboxCapPairs` resolves
|
||||
// the same record to find the private store), so there is no second half to attempt.
|
||||
return { inboxes: out, incomplete: { error } };
|
||||
return { inboxes: out, incomplete: { error, register: "account-record" } };
|
||||
}
|
||||
try {
|
||||
for (const { inbox } of await readInboxCapPairs()) {
|
||||
@@ -546,7 +565,7 @@ export async function enumerateMyInboxes(): Promise<InboxEnumeration> {
|
||||
out.push(inbox);
|
||||
}
|
||||
} catch (error) {
|
||||
return { inboxes: out, incomplete: { error } };
|
||||
return { inboxes: out, incomplete: { error, register: "user-branch" } };
|
||||
}
|
||||
return { inboxes: out, incomplete: null };
|
||||
}
|
||||
|
||||
@@ -57,7 +57,7 @@
|
||||
|
||||
import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap";
|
||||
import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log";
|
||||
import { enumerateMyInboxes, type InboxEnumeration } from "./branch-registers";
|
||||
import { enumerateMyInboxes, type InboxEnumeration, type InboxRegister } from "./branch-registers";
|
||||
import { drainInboxSerially, drainsSettled } from "./inbox-drain";
|
||||
import { lookupAccount } from "../shared-wallet/account-registry";
|
||||
import { processInbox } from "../surface/inbox";
|
||||
@@ -83,7 +83,7 @@ interface Observation {
|
||||
/** A trigger that arrived mid-cycle: the running one repeats once rather than lose it. */
|
||||
enumerateAgain: boolean;
|
||||
/**
|
||||
* True while the shortfall reported by the last enumeration is still the SAME one — so a
|
||||
* WHICH shortfall was reported last, or `null` when the last list came back WHOLE — so a
|
||||
* register that stays unreadable is reported once, not once per cycle.
|
||||
*
|
||||
* Every signal this observation listens to runs a full cycle, and a persistent shortfall
|
||||
@@ -91,8 +91,17 @@ interface Observation {
|
||||
* is nudged: one deposit produced three copies of the same line. That is what buries the
|
||||
* report that matters under the report that repeats. Cleared the moment a list comes back
|
||||
* WHOLE, so a shortfall that goes away and returns is a new occurrence and says so.
|
||||
*
|
||||
* ── The register, and not a bare "already said something" ─────────────────
|
||||
* There are TWO shortfalls, they are disjoint, and one is far worse than the other: the
|
||||
* account record failing leaves this identity with NOTHING watched, its own two inboxes
|
||||
* included, while the User branch failing leaves those two watched and only the
|
||||
* per-document inboxes missing ({@link InboxEnumeration.incomplete}). A boolean here made
|
||||
* the second silence the first — the condition CHANGED, to a materially different one, and
|
||||
* the log said nothing because something had already been said. Keyed by register, a
|
||||
* change of shortfall is what it is: news.
|
||||
*/
|
||||
shortfallReported: boolean;
|
||||
shortfallReported: InboxRegister | null;
|
||||
/** Enumerations and applications in flight — what {@link observationSettled} waits on. */
|
||||
pending: Set<Promise<void>>;
|
||||
}
|
||||
@@ -116,6 +125,22 @@ function reportUnobserved(what: string, error: unknown): void {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* What a shortfall on `register` leaves unwatched, in the words the log uses.
|
||||
*
|
||||
* The two lines have to READ differently, not merely be counted differently: a reader who
|
||||
* sees the second one after the first has to be able to tell that the condition changed, and
|
||||
* two identical sentences make an escalation look like a repeat. Both keep the phrase the
|
||||
* report has always ended on, because it is the one thing this line is scanned for.
|
||||
*/
|
||||
function shortfallSaid(register: InboxRegister): string {
|
||||
return register === "account-record"
|
||||
? "the inboxes to watch could not all be listed (the account record — your own two " +
|
||||
"inboxes, where a share addressed to you by name lands, are NOT among the ones watched)"
|
||||
: "the inboxes to watch could not all be listed (the register of the inboxes opened on " +
|
||||
"documents — your own two are watched)";
|
||||
}
|
||||
|
||||
/** Is `obs` still the live observation, for the identity it belongs to? */
|
||||
function current(obs: Observation): boolean {
|
||||
return observation === obs && getCurrentUser() === obs.holder;
|
||||
@@ -193,7 +218,10 @@ async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise<void> {
|
||||
* Once per OCCURRENCE, though, not once per enumeration — see
|
||||
* {@link Observation.shortfallReported}. The retry that repairs it is also what re-reads it,
|
||||
* so a condition that persists is re-discovered by every signal; reporting each discovery
|
||||
* says "it happened again" about the single thing that never stopped happening.
|
||||
* says "it happened again" about the single thing that never stopped happening. An
|
||||
* occurrence is per REGISTER, because the two shortfalls are two conditions: the list going
|
||||
* from "missing the document inboxes" to "missing your own two" is not the same fault
|
||||
* continuing, it is a worse one starting.
|
||||
*/
|
||||
async function watchTheInboxes(obs: Observation): Promise<void> {
|
||||
if (!current(obs)) return;
|
||||
@@ -207,10 +235,10 @@ async function watchTheInboxes(obs: Observation): Promise<void> {
|
||||
}
|
||||
if (!current(obs)) return;
|
||||
if (listed.incomplete === null) {
|
||||
obs.shortfallReported = false;
|
||||
} else if (!obs.shortfallReported) {
|
||||
obs.shortfallReported = true;
|
||||
reportUnobserved("the inboxes to watch could not all be listed", listed.incomplete.error);
|
||||
obs.shortfallReported = null;
|
||||
} else if (obs.shortfallReported !== listed.incomplete.register) {
|
||||
obs.shortfallReported = listed.incomplete.register;
|
||||
reportUnobserved(shortfallSaid(listed.incomplete.register), listed.incomplete.error);
|
||||
}
|
||||
for (const inbox of listed.inboxes) {
|
||||
if (obs.inboxes.has(inbox)) continue;
|
||||
@@ -373,7 +401,7 @@ export async function startObservingInboxes(): Promise<void> {
|
||||
holdings: null,
|
||||
enumerating: false,
|
||||
enumerateAgain: false,
|
||||
shortfallReported: false,
|
||||
shortfallReported: null,
|
||||
pending: new Set(),
|
||||
};
|
||||
observation = obs;
|
||||
|
||||
@@ -202,6 +202,15 @@ interface DocFanOut {
|
||||
lastState: { resp: DocChange; type: DocChangeType } | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Every document with at least one local listener.
|
||||
*
|
||||
* An entry in this map ALWAYS has listeners — that is the invariant the rest of the module
|
||||
* reads "is this fan-out still standing" off: a listener is added synchronously with the
|
||||
* entry that holds it ({@link subscribeDocUnguarded}), and the last one out takes the entry
|
||||
* with it ({@link releaseFanOut}, {@link resetDocSubscriptions}). So no caller asks after the
|
||||
* listener count separately; `fanOuts.get(nuri) === entry` already answers it.
|
||||
*/
|
||||
const fanOuts = new Map<Nuri, DocFanOut>();
|
||||
|
||||
/** Hand one push to one listener, isolating a throwing handler from the others. */
|
||||
@@ -333,15 +342,46 @@ function releaseFanOut(nuri: Nuri, entry: DocFanOut): void {
|
||||
* silence: `resubscribeDocs` runs on a session change, the superseded call sits on the
|
||||
* previous session's verifier, and nothing else will ever close it.
|
||||
*
|
||||
* ── Why the guard sits BEFORE the call, and not only after it ──────────────
|
||||
* It is what makes the sentence above true rather than merely likely. Superseded while the
|
||||
* session id was still resolving, this opens NO channel at all — so every attempt that owns
|
||||
* one issued its `doc_subscribe` strictly before the current attempt existed, and by the
|
||||
* global broker lock held for the whole call (`sdk/rust/src/local_broker.rs:3057`) it was
|
||||
* served first. Whoever inserts last wins the branch (`verifier.rs:361`), so a superseded
|
||||
* attempt is always the upstream LOSER, never the live channel. Without this guard the two
|
||||
* calls could reach the broker in either order, and releasing the loser would be a coin flip
|
||||
* on whether the document goes silent.
|
||||
* ── Why the guard sits BEFORE the call, and what it does NOT settle ────────
|
||||
* What it settles is a fact: superseded while the session id was still resolving, this opens
|
||||
* NO channel at all. That is the wider half of the window, and it closes it outright.
|
||||
*
|
||||
* What it leaves open is a BET, and it is written down here as one. For an attempt that DID
|
||||
* place its call before being superseded, this code releases the channel that call opened —
|
||||
* which is only ever releasing the upstream LOSER if the broker SERVED the two calls in the
|
||||
* order they were ISSUED. Issuing is all this module can order; service is the broker's, and
|
||||
* nothing found upstream promises the two agree:
|
||||
*
|
||||
* - `LOCAL_BROKER` is behind an `async_std::sync::RwLock` held for the whole call
|
||||
* (`sdk/rust/src/local_broker.rs:3057`), and that type IS `async_lock::RwLock`
|
||||
* (async-std 1.13.2 `src/sync/mod.rs:181`, a re-export);
|
||||
* - `RwLock::write()` takes the lock's internal `async_lock::Mutex` first (async-lock 3.4.1
|
||||
* `src/rwlock/raw.rs:163-168`), and that mutex is documented as "eventual fairness" —
|
||||
* fair ON AVERAGE, not FIFO (`src/mutex.rs:22-24`). Its hot loop re-runs a bare
|
||||
* `compare_exchange` on every poll, so a call arriving later can take the lock ahead of
|
||||
* one already waiting;
|
||||
* - and the anti-starvation fallback that would eventually force fairness is
|
||||
* `#[cfg(all(feature = "std", not(target_family = "wasm")))]` (`src/mutex.rs:578-581`) —
|
||||
* compiled out on wasm, which is where this SDK runs.
|
||||
*
|
||||
* So the lock SERIALISES the calls; it does not ORDER them. Two attempts on the same branch
|
||||
* of the same verifier is a reachable state — the last listener leaving while an establish is
|
||||
* in flight releases the fan-out, and the next joiner opens a fresh one against the same
|
||||
* session — so this is not a cross-session-only concern.
|
||||
*
|
||||
* What breaks the bet is therefore a service order inverted relative to the issue order, and
|
||||
* this is what it costs when it happens: the superseded call inserts LAST, so it wins the
|
||||
* branch and closes the current attempt's sender (`verifier.rs:361`); this code then releases
|
||||
* its own channel, leaving the branch holding a closed sender that `push_app_response` drops
|
||||
* at the first push (`verifier.rs:258-261`). Nothing rejects and nothing logs. The fan-out
|
||||
* holds a `realUnsub` and believes itself subscribed, `establishing` stays true so no later
|
||||
* joiner re-opens it, and the document is silently dead for the rest of the session —
|
||||
* `test/subscribe.test.ts` drives exactly that and pins it.
|
||||
*
|
||||
* Releasing is still what to do: DROPPING the superseded unsubscribe instead leaks a live
|
||||
* subscription in the ordinary case, and buys nothing in the inverted one — the winner's own
|
||||
* callback is gated on an attempt that is no longer current, so it delivers to nobody either
|
||||
* way. The bet is on which failure mode is reachable, never on releasing being free.
|
||||
*/
|
||||
async function establish(
|
||||
nuri: Nuri,
|
||||
@@ -426,7 +466,6 @@ export function resubscribeDocs(): void {
|
||||
return;
|
||||
}
|
||||
for (const [nuri, entry] of [...fanOuts]) {
|
||||
if (entry.listeners.size === 0) continue;
|
||||
const stale = entry.realUnsub;
|
||||
entry.realUnsub = null;
|
||||
// The barrier belonged to the session that is gone. Whoever joins next waits for a
|
||||
|
||||
Reference in New Issue
Block a user