fix: une barriere appartient a la tentative qui l'a produite
Seconde revue adverse, sur le correctif precedent - et elle a raison : celui-ci MASQUAIT la barriere perimee au lieu de l'oublier. `reportSetupFailure` effacait `establishing` et laissait `lastState`. Le premier correctif avait ajoute « et un canal tient » du cote du LECTEUR, ce qui ne cachait la barriere morte que tant que rien ne tenait. La souscription suivante sur le meme fan-out remettait `establishing` a vrai, le masque tombait - et le fan-out rejouait ce `State` au nouvel arrivant. Mesure : l'ouverture bootstrap franchissait sa barriere en 2 ms sur un `State` d'un appel qui avait ECHOUE, journalisait `BARRIER ... synced`, et `getSyncState` repondait `"synced"` - donc autorite, donc `noRowsMeansAbsent` prenait une lecture a zero ligne pour une absence definitive. Le depot detruit que le commit precedent pretendait empecher. Corrige a la source : une tentative qui echoue oublie sa barriere, comme le font deja `releaseFanOut` et `resubscribeDocs`. Le garde du lecteur est RETIRE avec, et pas garde en ceinture-bretelles : il etait devenu inatteignable, et c'est lui qui avait cache la faute. Apres : la barriere est oubliee, l'ouverture attend la sienne (202 ms), `"timed-out"`. Et le test que j'avais ecrit affirmait le defaut - `barrierCrossed` vrai - au lieu de l'etat correct. Corrige, plus un second test qui conduit le rejeu de bout en bout.
This commit is contained in:
@@ -300,19 +300,20 @@ 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
|
// @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" {
|
export function getSyncState(nuri: Nuri): SyncState | "unknown" {
|
||||||
const channel = docChannel(nuri);
|
const channel = docChannel(nuri);
|
||||||
// The first `State` on the channel STANDING NOW - both halves, and the second was missing in
|
// The first `State` on the channel standing NOW. {@link docChannel} guarantees both halves -
|
||||||
// this repair's first draft. `barrierCrossed` alone said `"synced"` for a fan-out whose
|
// it does not report a barrier while no channel stands - so this asks the one question, and
|
||||||
// establish had been REFUSED after pushing a `State`: `reportSetupFailure` clears
|
// that is deliberate: a first draft of this repair re-imposed the second half HERE instead,
|
||||||
// `establishing` and leaves `lastState`, so the barrier outlived the attempt that produced it,
|
// which only MASKED a stale barrier rather than forgetting it. The next subscribe on the same
|
||||||
// and `"synced"` is AUTHORITY - `index-deposit.ts` `noRowsMeansAbsent` then calls a zero-row
|
// fan-out brought it straight back, was replayed it, crossed its own barrier in 2ms on a
|
||||||
// read a definitive absence. The previous registry answered `"timed-out"` there, and strict is
|
// `State` from a call that had FAILED, and answered `"synced"` - which is AUTHORITY, so
|
||||||
// the reading this family is built on: the cost of it is latency, never loss.
|
// `index-deposit.ts` `noRowsMeansAbsent` called a zero-row read a definitive absence and the
|
||||||
|
// deposit was destroyed. A guard at the reader cannot fix a fact that is wrong at the source.
|
||||||
//
|
//
|
||||||
// Asked of the fan-out rather than remembered, because that is the one record of it: a barrier
|
// Asked of the fan-out rather than remembered, because that is the one record of it: a barrier
|
||||||
// is forgotten when its channel is released and when its session changes. Answered even for a
|
// is forgotten when its attempt fails, when its channel is released and when its session
|
||||||
// document this module never opened - an application's own subscription crosses the same
|
// changes. Answered even for a document this module never opened - an application's own
|
||||||
// barrier, and `"synced"` is then simply true.
|
// subscription crosses the same barrier, and `"synced"` is then simply true.
|
||||||
if (channel.state !== "none" && channel.barrierCrossed) return "synced";
|
if (channel.barrierCrossed) return "synced";
|
||||||
const open = opens.get(nuri);
|
const open = opens.get(nuri);
|
||||||
// Never opened here, or opened on the no-`doc_subscribe` path, which records nothing
|
// Never opened here, or opened on the no-`doc_subscribe` path, which records nothing
|
||||||
// because there is no barrier to have missed.
|
// because there is no barrier to have missed.
|
||||||
|
|||||||
@@ -299,6 +299,14 @@ export function docChannel(nuri: Nuri): DocChannel {
|
|||||||
// two together are the only thing that separates "asked" from "open" - and recording the
|
// two together are the only thing that separates "asked" from "open" - and recording the
|
||||||
// first as if it were the second is the defect this accessor exists to make impossible.
|
// first as if it were the second is the defect this accessor exists to make impossible.
|
||||||
const state: DocChannelState = !entry.establishing ? "none" : entry.realUnsub !== null ? "open" : "opening";
|
const state: DocChannelState = !entry.establishing ? "none" : entry.realUnsub !== null ? "open" : "opening";
|
||||||
|
// `lastState` alone, and deliberately NOT `&& state !== "none"`. Every path that stops an
|
||||||
|
// attempt standing forgets the barrier with it - {@link releaseFanOut}, {@link resubscribeDocs}
|
||||||
|
// and, since this was found the hard way, {@link reportSetupFailure} - so the two can no longer
|
||||||
|
// disagree, and a guard here would be unreachable. It would also be a MASK: the first repair of
|
||||||
|
// that defect put exactly this condition in the reader instead of forgetting the barrier at the
|
||||||
|
// source, and it hid the fault while the fan-out went on replaying a dead `State` to the next
|
||||||
|
// joiner. A fact that is wrong at the source has to be fixed at the source; a reader that
|
||||||
|
// papers over it only moves the lie.
|
||||||
return { state, barrierCrossed: entry.lastState !== null, incarnation: entry.incarnation };
|
return { state, barrierCrossed: entry.lastState !== null, incarnation: entry.incarnation };
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -383,6 +391,15 @@ function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown, attemp
|
|||||||
console.error("[subscribe] doc_subscribe failed for", nuri, error);
|
console.error("[subscribe] doc_subscribe failed for", nuri, error);
|
||||||
if (!isCurrentAttempt(nuri, entry, attempt)) return;
|
if (!isCurrentAttempt(nuri, entry, attempt)) return;
|
||||||
entry.establishing = false;
|
entry.establishing = false;
|
||||||
|
// A BARRIER BELONGS TO THE ATTEMPT THAT PRODUCED IT, and this one failed. The push may well
|
||||||
|
// have arrived - the callback is driven locally while the call's own answer travels the RPC -
|
||||||
|
// but the call did not take, so there is no subscription behind that `State` and nothing will
|
||||||
|
// follow it. Left here, the fan-out replayed it to the next joiner: a bootstrap open crossed
|
||||||
|
// its barrier in 2ms on a `State` from a call that had FAILED, logged `BARRIER ... synced`,
|
||||||
|
// and `getSyncState` answered `"synced"` - which is authority, so a zero-row read over the
|
||||||
|
// document became a definitive absence and a deposit was destroyed. {@link releaseFanOut}
|
||||||
|
// and {@link resubscribeDocs} both forget it for the same reason; this path did not.
|
||||||
|
entry.lastState = null;
|
||||||
for (const listener of [...entry.listeners]) {
|
for (const listener of [...entry.listeners]) {
|
||||||
const tell = listener.onSetupFailed;
|
const tell = listener.onSetupFailed;
|
||||||
if (tell === null || !entry.listeners.has(listener)) continue;
|
if (tell === null || !entry.listeners.has(listener)) continue;
|
||||||
|
|||||||
@@ -367,15 +367,15 @@ describe("ensureRepoOpen", () => {
|
|||||||
expect(getSyncState(doc)).toBe("synced");
|
expect(getSyncState(doc)).toBe("synced");
|
||||||
});
|
});
|
||||||
|
|
||||||
// A barrier must not outlive the attempt that produced it.
|
// A BARRIER BELONGS TO THE ATTEMPT THAT PRODUCED IT - and must be FORGOTTEN when that attempt
|
||||||
|
// fails, not merely ignored while nothing stands.
|
||||||
//
|
//
|
||||||
// `subscribe.ts` clears `establishing` on a refused establish and LEAVES the remembered
|
// The push can arrive and the call still fail: the callback is driven locally while the call's
|
||||||
// `State`, so a call that pushed the barrier and then failed leaves the fan-out with a
|
// own answer travels the RPC. `subscribe.ts` cleared `establishing` on a refused establish and
|
||||||
// barrier and no channel. Reading `"synced"` off that is AUTHORITY - `index-deposit.ts`
|
// LEFT the remembered `State`, so the fan-out held a barrier with no subscription behind it.
|
||||||
// `noRowsMeansAbsent` calls a zero-row read over it a definitive absence, which destroys a
|
// The first repair only made `getSyncState` ignore it while the channel read `"none"`, which
|
||||||
// deposit - and the registry this replaced answered `"timed-out"` there. Strict is the
|
// MASKED it - see the next test for what the mask let through.
|
||||||
// reading: the cost of it is latency, never loss.
|
it("a barrier pushed by a subscribe that then FAILED is forgotten, not merely ignored", async () => {
|
||||||
it("a barrier pushed by a subscribe that then FAILED is not authority", async () => {
|
|
||||||
const doc_subscribe = mock(async (_n: string, _s: string, cb: (r: unknown) => void) => {
|
const doc_subscribe = mock(async (_n: string, _s: string, cb: (r: unknown) => void) => {
|
||||||
cb({ V0: { State: {} } }); // the barrier arrives...
|
cb({ V0: { State: {} } }); // the barrier arrives...
|
||||||
throw new Error("RepoNotFound: after the push"); // ...and the call then fails
|
throw new Error("RepoNotFound: after the push"); // ...and the call then fails
|
||||||
@@ -390,18 +390,62 @@ describe("ensureRepoOpen", () => {
|
|||||||
setOpenTimeoutForTests(20);
|
setOpenTimeoutForTests(20);
|
||||||
const doc = "did:ng:o:pushed-then-failed";
|
const doc = "did:ng:o:pushed-then-failed";
|
||||||
|
|
||||||
// Something else holds the fan-out, so the refusal does not take the entry with it and the
|
// Something else holds the fan-out, so the refusal does not take the entry with it - which is
|
||||||
// remembered barrier survives the attempt that pushed it.
|
// what would have cleared the barrier for the wrong reason.
|
||||||
const other = subscribeDoc(doc, () => {});
|
const other = subscribeDoc(doc, () => {});
|
||||||
await ensureRepoOpen(doc);
|
await ensureRepoOpen(doc);
|
||||||
await new Promise((r) => setTimeout(r, 10));
|
await new Promise((r) => setTimeout(r, 10));
|
||||||
|
|
||||||
expect(docChannel(doc).barrierCrossed).toBe(true); // the barrier IS remembered...
|
expect(docChannel(doc).barrierCrossed).toBe(false); // forgotten with the attempt
|
||||||
expect(docChannel(doc).state).toBe("none"); // ...and nothing stands behind it
|
expect(docChannel(doc).state).toBe("none");
|
||||||
expect(getSyncState(doc)).toBe("timed-out");
|
expect(getSyncState(doc)).toBe("timed-out");
|
||||||
other();
|
other();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// What MASKING the stale barrier let through, and it is the destroyed deposit this whole family
|
||||||
|
// exists to prevent.
|
||||||
|
//
|
||||||
|
// A fan-out that kept the `State` of a FAILED call replays it to the next joiner. So the next
|
||||||
|
// subscribe restored `establishing`, the channel left `"none"`, the mask lifted - and a
|
||||||
|
// bootstrap open crossed its barrier in two milliseconds on a `State` from a call that had
|
||||||
|
// never taken, logging `BARRIER ... synced`. `getSyncState` then answers `"synced"`, which is
|
||||||
|
// AUTHORITY: `index-deposit.ts` `noRowsMeansAbsent` calls a zero-row read a definitive absence.
|
||||||
|
it("the next subscribe is not handed the barrier of a call that FAILED", async () => {
|
||||||
|
let calls = 0;
|
||||||
|
const doc_subscribe = mock(async (_n: string, _s: string, cb: (r: unknown) => void) => {
|
||||||
|
calls += 1;
|
||||||
|
if (calls === 1) {
|
||||||
|
cb({ V0: { State: {} } });
|
||||||
|
throw new Error("RepoNotFound: after the push");
|
||||||
|
}
|
||||||
|
return await new Promise<never>(() => {}); // the retry never answers
|
||||||
|
});
|
||||||
|
const ng = {
|
||||||
|
doc_subscribe,
|
||||||
|
sparql_query: mock(async () => ({ results: { bindings: [] } })),
|
||||||
|
doc_create: mock(async () => "did:ng:o:new"),
|
||||||
|
sparql_update: mock(async () => undefined),
|
||||||
|
};
|
||||||
|
inject(ng as unknown as ReturnType<typeof makeFakeNgWithSubscribe>);
|
||||||
|
setOpenTimeoutForTests(200);
|
||||||
|
const doc = "did:ng:o:replayed-corpse";
|
||||||
|
|
||||||
|
const app = subscribeDoc(doc, () => {}); // the application's own subscription fails
|
||||||
|
await new Promise((r) => setTimeout(r, 20));
|
||||||
|
|
||||||
|
// The bootstrap open joins the SAME fan-out and starts a second establish.
|
||||||
|
const at = Date.now();
|
||||||
|
await ensureRepoOpen(doc);
|
||||||
|
const paid = Date.now() - at;
|
||||||
|
|
||||||
|
expect(calls).toBe(2); // a real second attempt was made...
|
||||||
|
expect(paid).toBeGreaterThanOrEqual(150); // ...and this open waited its OWN barrier out
|
||||||
|
expect(docChannel(doc).barrierCrossed).toBe(false);
|
||||||
|
expect(getSyncState(doc)).toBe("timed-out"); // never authority over a corpse
|
||||||
|
app();
|
||||||
|
resetOpenedRepos();
|
||||||
|
});
|
||||||
|
|
||||||
// `release` does not answer "am I still listening", and treating it as though it did leaves the
|
// `release` does not answer "am I still listening", and treating it as though it did leaves the
|
||||||
// document held by somebody else's listener.
|
// document held by somebody else's listener.
|
||||||
//
|
//
|
||||||
|
|||||||
Reference in New Issue
Block a user