Sylvain Duchesne
0a2b8b1fbf
Le contrat publie et l'api-contract etaient traites avec le retrait lui-meme ; restaient les documents qui decrivaient readUnion comme la primitive de lecture du paquet, ou qui nommaient le module disparu. Ce que les lecteurs apprennent maintenant : lire un document, c'est docs.sparqlQuery ancre dessus, et lire plusieurs documents, c'est cet appel une fois par document -- avec la decision sur une lecture qui echoue ecrite au point d'appel. Deux consequences sont dites plutot que decouvertes : une lecture refusee leve au lieu de rendre une liste vide, et les sujets sous le prefixe reserve urn:ng-eventually: reviennent dans le resultat, a l'appelant de les ignorer -- sans aucun gain de droits, les deux registres qu'un document ordinaire porte etant l'adresse de sa boite et, pour un document public, la cle que son magasin sert deja a qui la demande. Les briefs dates ne sont pas touches : ils decrivent un etat de juillet, et le reecrire falsifierait ce qui avait ete decide alors.
635 lines
27 KiB
TypeScript
635 lines
27 KiB
TypeScript
/**
|
|
* Applying an inbox deposit is a DISPATCH on what the deposit declares itself to be, and
|
|
* this suite is about the two things a `match` owes: every arm does real work, and the
|
|
* arm that does not exist SAYS SO.
|
|
*
|
|
* ── What each test would have looked like before ──────────────────────────
|
|
* Before 2026-08-21 `processInbox` iterated the caps a read had observed. A deposit of any
|
|
* other shape was read, counted, logged — and produced nothing: no effect, and no reported
|
|
* failure. So a reference deposit vanished, a malformed Link vanished, and a kind nobody
|
|
* had written an arm for vanished, all three indistinguishable from an empty inbox. Every
|
|
* test below fails against that code, and fails again if the arm it exercises is removed.
|
|
*
|
|
* ── Nobody is handed an inbox address ─────────────────────────────────────
|
|
* Each depositor names the DOCUMENT (`inbox.postToDocument`), which is all an application
|
|
* has; the address is resolved inside. The one address the TEST resolves for itself is
|
|
* used only to assert what landed, never given to an actor.
|
|
*/
|
|
import { test, expect, mock, afterEach } from "bun:test";
|
|
import { configure } from "../src/index";
|
|
import {
|
|
configureStoreRegistry,
|
|
resetCaps,
|
|
resetConfig,
|
|
resetStoreRegistry,
|
|
setCurrentUser,
|
|
} from "../src/shared-wallet/bootstrap";
|
|
import {
|
|
createEntityDoc,
|
|
resetRegistryCache,
|
|
resolveWriteGraph,
|
|
} from "../src/shared-wallet/account-registry";
|
|
import type { RegistrySession } from "../src/shared-wallet/account-registry";
|
|
import { openDocumentInbox } from "../src/emulated-verifier/branch-registers";
|
|
import { connectedUser } from "../src/emulated-verifier/connect";
|
|
import {
|
|
cancelScheduledInboxProcessing,
|
|
runScheduledInboxProcessingNow,
|
|
} from "../src/emulated-verifier/inbox-processor";
|
|
import { stopObservingInboxes } from "../src/emulated-verifier/inbox-observer";
|
|
import { resetOpenedRepos, setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
|
|
import { resetPublicStoreFetches } from "../src/emulated-verifier/public-store";
|
|
import { ENTRY_VALUE, INDEX_FIELD } from "../src/emulated-verifier/index-deposit";
|
|
import { postToDocument } from "../src/surface/inbox";
|
|
import { readDocumentSubjects } from "../src/surface/read-document";
|
|
import { sparqlUpdate } from "../src/surface/docs";
|
|
import type { Nuri } from "../src/model/types";
|
|
|
|
const SESSION: RegistrySession = { sessionId: "sid-dispatch", privateStoreId: "PRIV-DISPATCH" };
|
|
const SHIM = "urn:ng-eventually:shim";
|
|
const INBOX = "urn:ng-eventually:inbox";
|
|
/** The field the index documents below declare they index by. */
|
|
const WHEN = "urn:test:when";
|
|
|
|
interface Quad { g: string; s: string; p: string; o: string }
|
|
|
|
function unescapeLiteral(s: string): string {
|
|
let out = "";
|
|
for (let i = 0; i < s.length; i++) {
|
|
if (s[i] === "\\" && i + 1 < s.length) {
|
|
const next = s[++i];
|
|
out += next === "n" ? "\n" : next === "r" ? "\r" : next === "t" ? "\t" : next!;
|
|
} else out += s[i];
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/** A stateful fake `ng`: the shim SPARQL, the inbox SPARQL, and the anchored per-doc read. */
|
|
function makeFakeNg() {
|
|
const quads: Quad[] = [];
|
|
let docCounter = 0;
|
|
|
|
const doc_create = mock(async () => `did:ng:o:doc${++docCounter}`);
|
|
|
|
const sparql_update = mock(async (...a: unknown[]) => {
|
|
const query = a[1] as string;
|
|
const anchor = a[2] as string | undefined;
|
|
if (!anchor) return undefined;
|
|
const del = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+<([^>]+)>\s+\?/);
|
|
if (del) {
|
|
const [s0, p0] = [del[1]!, del[2]!];
|
|
for (let i = quads.length - 1; i >= 0; i--) {
|
|
const q = quads[i]!;
|
|
if (q.g === anchor && q.s === s0 && q.p === p0) quads.splice(i, 1);
|
|
}
|
|
return undefined;
|
|
}
|
|
// Retiring a DEPOSIT: every triple of one subject, whatever predicates it carries
|
|
// (`emulated-verifier/inbox-consume.removeDeposit`). Modelled here rather than ignored
|
|
// — a fake that dropped this delete would leave every deposit in place and let this
|
|
// whole suite pass on a queue that never shrinks.
|
|
const delAll = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+\?\w+\s+\?\w+\s*\}/);
|
|
if (delAll) {
|
|
const s0 = delAll[1]!;
|
|
for (let i = quads.length - 1; i >= 0; i--) {
|
|
const q = quads[i]!;
|
|
if (q.g === anchor && q.s === s0) quads.splice(i, 1);
|
|
}
|
|
return undefined;
|
|
}
|
|
const body = query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, "");
|
|
const sm = body.match(/<([^>]+)>/);
|
|
if (!sm) return undefined;
|
|
const s = sm[1]!;
|
|
const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g;
|
|
let m: RegExpExecArray | null;
|
|
const after = body.slice(body.indexOf(sm[0]) + sm[0].length);
|
|
while ((m = pairRe.exec(after)) !== null) {
|
|
const p = m[1] ?? (query.includes(`${INBOX}:Deposit`) ? `${INBOX}:Deposit` : `${SHIM}:Account`);
|
|
const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? "");
|
|
quads.push({ g: anchor, s, p, o });
|
|
}
|
|
return undefined;
|
|
});
|
|
|
|
const sparql_query = mock(async (...a: unknown[]) => {
|
|
const query = a[1] as string;
|
|
const anchor = a[3] as string | undefined;
|
|
const byPred = (pred: string, v: string) => ({
|
|
results: {
|
|
bindings: quads
|
|
.filter((q) => q.g === anchor && q.p === pred)
|
|
.map((q) => ({ [v]: { value: q.o } })),
|
|
},
|
|
});
|
|
if (query.includes(`<${SHIM}:shimDoc>`)) return byPred(`${SHIM}:shimDoc`, "shimDoc");
|
|
if (query.includes(`<${SHIM}:id>`)) {
|
|
const subjM = query.match(/<([^>]+)>\s+a\s+<urn:ng-eventually:shim:Account>/);
|
|
const only = subjM ? subjM[1]! : null;
|
|
const bySubject = new Map<string, Record<string, string>>();
|
|
for (const q of quads) {
|
|
if (q.g !== anchor) continue;
|
|
if (only !== null && q.s !== only) continue;
|
|
const rec = bySubject.get(q.s) ?? {};
|
|
if (q.p === `${SHIM}:id`) rec.id = q.o;
|
|
if (q.p === `${SHIM}:docPublic`) rec.docPublic = q.o;
|
|
if (q.p === `${SHIM}:docProtected`) rec.docProtected = q.o;
|
|
if (q.p === `${SHIM}:docPrivate`) rec.docPrivate = q.o;
|
|
bySubject.set(q.s, rec);
|
|
}
|
|
return {
|
|
results: {
|
|
bindings: [...bySubject.values()].filter((r) => r.id).map((r) => ({
|
|
id: { value: r.id! },
|
|
docPublic: { value: r.docPublic ?? "" },
|
|
docProtected: { value: r.docProtected ?? "" },
|
|
docPrivate: { value: r.docPrivate ?? "" },
|
|
})),
|
|
},
|
|
};
|
|
}
|
|
if (query.includes(`<${INBOX}:payload>`)) {
|
|
const bySubject = new Map<string, Record<string, string>>();
|
|
for (const q of quads) {
|
|
if (q.g !== anchor) continue;
|
|
const rec = bySubject.get(q.s) ?? {};
|
|
// The deposit's own subject — the key a drain names to retire it.
|
|
rec.d = q.s;
|
|
if (q.p === `${INBOX}:payload`) rec.payload = q.o;
|
|
if (q.p === `${INBOX}:ts`) rec.ts = q.o;
|
|
if (q.p === `${INBOX}:from`) rec.from = q.o;
|
|
bySubject.set(q.s, rec);
|
|
}
|
|
return {
|
|
results: {
|
|
bindings: [...bySubject.values()]
|
|
.filter((r) => r.payload !== undefined && r.ts !== undefined)
|
|
.map((r) => {
|
|
const row: Record<string, { value: string }> = {
|
|
d: { value: r.d! },
|
|
payload: { value: r.payload! },
|
|
ts: { value: r.ts! },
|
|
};
|
|
if (r.from !== undefined) row.from = { value: r.from };
|
|
return row;
|
|
}),
|
|
},
|
|
};
|
|
}
|
|
if (query.includes(`<${SHIM}:inboxCap>`)) return byPred(`${SHIM}:inboxCap`, "c");
|
|
if (query.includes(`<${SHIM}:inboxAddress>`)) return byPred(`${SHIM}:inboxAddress`, "a");
|
|
if (query.includes(`<${SHIM}:readCap>`)) return byPred(`${SHIM}:readCap`, "c");
|
|
if (query.includes(`<${SHIM}:link>`)) return byPred(`${SHIM}:link`, "c");
|
|
if (query.includes(`${SHIM}:isInbox`)) return byPred(`${SHIM}:isInbox`, "i");
|
|
if (query.includes(`<${SHIM}:inboxOwner>`)) {
|
|
const sm = query.match(/<([^>]+)>\s+<urn:ng-eventually:shim:inboxOwner>/);
|
|
const subj = sm ? sm[1]! : null;
|
|
return {
|
|
results: {
|
|
bindings: quads
|
|
.filter((q) => q.g === anchor && q.p === `${SHIM}:inboxOwner` && q.s === subj)
|
|
.map((q) => ({ u: { value: q.o } })),
|
|
},
|
|
};
|
|
}
|
|
if (query.includes(`${SHIM}:docInbox`)) {
|
|
const pm = query.match(/<(urn:ng-eventually:shim:docInbox:[a-z]+)>/);
|
|
const pred = pm ? pm[1]! : "";
|
|
const sm = query.match(/<([^>]+)>\s+<urn:ng-eventually:shim:docInbox/);
|
|
const subj = sm ? sm[1]! : null;
|
|
return {
|
|
results: {
|
|
bindings: quads
|
|
.filter((q) => q.g === anchor && q.p === pred && (subj === null || q.s === subj))
|
|
.map((q) => ({ d: { value: q.o } })),
|
|
},
|
|
};
|
|
}
|
|
if (query.includes(`<${SHIM}:exposedReadCap>`)) return byPred(`${SHIM}:exposedReadCap`, "c");
|
|
if (query.includes(`<${SHIM}:contains>`)) return byPred(`${SHIM}:contains`, "e");
|
|
return {
|
|
results: {
|
|
bindings: quads
|
|
.filter((q) => q.g === anchor)
|
|
.map((q) => ({ s: { value: q.s }, p: { value: q.p }, o: { value: q.o } })),
|
|
},
|
|
};
|
|
});
|
|
|
|
// `doc_subscribe` is declared and left undefined: absent, `ensureRepoOpen` is the
|
|
// documented no-op of the unit-fake path. One test installs a subscription that never
|
|
// pushes `State`, to reach the `timed-out` repo state a cold start really produces.
|
|
const fakeNg: {
|
|
doc_create: typeof doc_create;
|
|
sparql_update: typeof sparql_update;
|
|
sparql_query: typeof sparql_query;
|
|
quads: Quad[];
|
|
doc_subscribe?: (...a: unknown[]) => () => void;
|
|
} = { doc_create, sparql_update, sparql_query, quads };
|
|
return fakeNg;
|
|
}
|
|
|
|
let fake: ReturnType<typeof makeFakeNg>;
|
|
|
|
function inject() {
|
|
fake = makeFakeNg();
|
|
configure({ ng: fake as never, useShape: (() => {}) as never });
|
|
configureStoreRegistry({ getSession: async () => SESSION });
|
|
resetRegistryCache();
|
|
resetOpenedRepos();
|
|
resetPublicStoreFetches();
|
|
resetCaps();
|
|
setCurrentUser(null);
|
|
}
|
|
|
|
afterEach(() => {
|
|
cancelScheduledInboxProcessing();
|
|
stopObservingInboxes();
|
|
resetConfig();
|
|
resetStoreRegistry();
|
|
resetCaps();
|
|
resetRegistryCache();
|
|
resetOpenedRepos();
|
|
resetPublicStoreFetches();
|
|
setCurrentUser(null);
|
|
});
|
|
|
|
/** Write one triple into `doc`, as the consumer's own write path would. */
|
|
async function write(doc: Nuri, p: string, o: string): Promise<void> {
|
|
await sparqlUpdate(SESSION.sessionId, `INSERT DATA { <${doc}> <${p}> "${o}" }`, doc, "test");
|
|
}
|
|
|
|
/** Every value the index holds for `object` — one per entry written. */
|
|
async function entryValuesFor(index: Nuri, object: Nuri): Promise<string[]> {
|
|
const subjects = await readDocumentSubjects(index);
|
|
return subjects.filter((s) => s.subject === object).flatMap((s) => s.props[ENTRY_VALUE] ?? []);
|
|
}
|
|
|
|
/** What the package reported, for the span of `run`. Restores `console.error` whatever
|
|
* happens — a suite that leaks a stub takes the next file down with it. */
|
|
async function reportedDuring(run: () => Promise<void>): Promise<string[]> {
|
|
const lines: string[] = [];
|
|
const real = console.error;
|
|
console.error = ((...args: unknown[]) => {
|
|
lines.push(args.map((a) => String(a)).join(" "));
|
|
}) as typeof console.error;
|
|
try {
|
|
await run();
|
|
} finally {
|
|
console.error = real;
|
|
}
|
|
return lines;
|
|
}
|
|
|
|
/**
|
|
* Alice's index: a public document declaring the field it indexes by, with an inbox open
|
|
* on it so that anyone can hand it a reference. Returns the index and one PUBLIC object
|
|
* carrying a value for that field.
|
|
*/
|
|
async function aliceOpensAnIndex(fields: string[]): Promise<{ index: Nuri; object: Nuri }> {
|
|
setCurrentUser("alice");
|
|
const index = await createEntityDoc("alice", "public");
|
|
for (const field of fields) await write(index, INDEX_FIELD, field);
|
|
await openDocumentInbox(index);
|
|
const object = await createEntityDoc("alice", "public");
|
|
await write(object, WHEN, "2026-01-01");
|
|
setCurrentUser(null);
|
|
return { index, object };
|
|
}
|
|
|
|
/** Bob has been in the page once, so he is someone the wallet knows. */
|
|
async function bobSignsInOnce(): Promise<void> {
|
|
setCurrentUser("bob");
|
|
await resolveWriteGraph("bob", "protected");
|
|
setCurrentUser(null);
|
|
}
|
|
|
|
/** Bob hands `payload` to the owner of `doc`, naming the document and nothing else. */
|
|
async function bobDepositsInto(doc: Nuri, payload: unknown, ts: number): Promise<void> {
|
|
setCurrentUser("bob");
|
|
await postToDocument(doc, { payload, ts });
|
|
setCurrentUser(null);
|
|
}
|
|
|
|
/** Alice connects — which restores her registers and drains every inbox she holds. */
|
|
async function aliceConnects(): Promise<void> {
|
|
setCurrentUser("alice");
|
|
await connectedUser();
|
|
}
|
|
|
|
// --- the reference arm -------------------------------------------------------
|
|
|
|
test("an entry appears in the index from a bare reference deposited by a stranger", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// The whole payload is the reference: no operation, no claim, and no copy of the value.
|
|
// What lands in the index is what the OBJECT says, which is why a stranger depositing
|
|
// achieves exactly what the owner would have.
|
|
await bobDepositsInto(index, object, 1);
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
|
|
await aliceConnects();
|
|
|
|
expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
|
|
});
|
|
|
|
test("a reference nobody can resolve is reported, and costs the index nothing", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// A reference this holder cannot reach — no cap, so the anchored read REFUSES it (until
|
|
// `1.0.0-dev.8` `readUnion` dropped it before issuing a query, which reached this arm as
|
|
// an empty answer instead of a failure) — and a resolvable one BEHIND it, which is what
|
|
// makes the question answerable at all.
|
|
await bobDepositsInto(index, "did:ng:o:nowhere", 1);
|
|
await bobDepositsInto(index, object, 2);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
// NOT called an absence, and this is the correction that cost the most to find. "No cap"
|
|
// reads as deterministic and is not: for a stranger's public document possession comes
|
|
// from a network fetch that swallows its errors and memoises the failure, so one outage
|
|
// presents exactly like a document that never existed. Classing it absent destroyed the
|
|
// deposit and every re-deposit after it. It stops the drain instead.
|
|
expect(reported.some((l) => /did:ng:o:nowhere/.test(l))).toBe(true);
|
|
expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
|
|
expect(await entryValuesFor(index, "did:ng:o:nowhere")).toEqual([]);
|
|
|
|
// The price is LATENCY, not loss: the unreadable one was popped, so the next connection
|
|
// resumes past it and the resolvable one lands. Nothing wedges, and nothing is destroyed
|
|
// that a working broker would have delivered.
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
await aliceConnects();
|
|
expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
|
|
});
|
|
|
|
test("an object already in the index is passed over — the same reference twice makes ONE entry", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// TWO deposits, each consumed when it is applied — not one deposit replayed. The check
|
|
// that a reference already in the index is passed over is what makes this land on ONE
|
|
// entry: without it the second deposit writes a second value for the same object, and an
|
|
// entry carrying two values is exactly what once made the index SHRINK through additions.
|
|
// That is why the check outlived consumption: it never answered the replay.
|
|
await bobDepositsInto(index, object, 1);
|
|
await bobDepositsInto(index, object, 2);
|
|
|
|
await aliceConnects();
|
|
|
|
expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
|
|
});
|
|
|
|
test("an index declaring NO field refuses, loudly, rather than writing something wrong", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
// A document with an inbox and no declaration: either it is not an index, or it could
|
|
// not be read — and here those are the same empty result.
|
|
const { index, object } = await aliceOpensAnIndex([]);
|
|
|
|
await bobDepositsInto(index, object, 1);
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /declares no index field/.test(l))).toBe(true);
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
});
|
|
|
|
test("an index declaring SEVERAL fields refuses rather than picking one", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
// Entries already written are never re-read, so curating under a second field would
|
|
// leave one list ordered by two different properties — a quiet wrong answer.
|
|
const { index, object } = await aliceOpensAnIndex([WHEN, "urn:test:other"]);
|
|
|
|
await bobDepositsInto(index, object, 1);
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /declares 2 index fields/.test(l))).toBe(true);
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
});
|
|
|
|
test("a reference deposit is left for its owner's own session, not applied from a stranger's", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
await bobDepositsInto(index, object, 1);
|
|
// The deferred stand-in for an absent owner runs under BOB's session. It files the
|
|
// verifier's own registers for an absent owner and stops there: writing application data
|
|
// into Alice's documents from Bob's page is a different act, and it is not taken.
|
|
setCurrentUser("bob");
|
|
const reported = await reportedDuring(async () => {
|
|
await runScheduledInboxProcessingNow();
|
|
});
|
|
setCurrentUser(null);
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
// Waiting is not failing, and it must not be reported as one: the arm did not try and
|
|
// fall short, it is not this session's to run. Running it anyway would report a refusal
|
|
// here every twenty seconds — Bob's User branch pairs no document with Alice's inbox.
|
|
expect(reported.filter((l) => /could not apply a deposit/.test(l))).toEqual([]);
|
|
|
|
// Nothing was consumed, so Alice's own next connection applies it — the real path.
|
|
await aliceConnects();
|
|
expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
|
|
});
|
|
|
|
// --- the cap arm -------------------------------------------------------------
|
|
|
|
test("a deposit declaring itself a Link and carrying no readable cap is reported, not passed over", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// A bare reference where a ReadCap belongs — the one type confusion that would invert
|
|
// the model if it were filed, and the shape a version skew produces. Upstream an arm
|
|
// opens by validating its envelope and a malformed one is a hard failure, not a skip.
|
|
await bobDepositsInto(index, { kind: "urn:ng-eventually:inbox:link", cap: "did:ng:o:bare" }, 1);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /declares itself a Link and carries no readable cap/.test(l))).toBe(true);
|
|
});
|
|
|
|
// --- the arm that does not exist ---------------------------------------------
|
|
|
|
test("a kind no arm answers says so — the dispatch's `NotImplemented`", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// A deposit declaring itself in THIS package's reserved namespace, for a variant nothing
|
|
// here applies. Upstream that is `_ => Err(NotImplemented)`; here it used to be silence.
|
|
await bobDepositsInto(index, { kind: "urn:ng-eventually:inbox:not-a-thing" }, 1);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(
|
|
reported.some((l) => /no arm applies deposits of kind "urn:ng-eventually:inbox:not-a-thing"/.test(l)),
|
|
).toBe(true);
|
|
});
|
|
|
|
test("a consumer payload is not a variant: no arm claims it and nothing is reported", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// `kind` is an ordinary word an application may use for its own messages. Keying the
|
|
// dispatch on the mere presence of the field would have turned every one of them into an
|
|
// unapplied message reported at each drain.
|
|
await bobDepositsInto(index, { kind: "join", who: "bob" }, 1);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.filter((l) => /could not apply a deposit/.test(l))).toEqual([]);
|
|
});
|
|
|
|
// --- a consumed deposit is never discarded in silence ------------------------
|
|
|
|
test("a reference to an object carrying nothing for the field is REPORTED, not silently dropped", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// A draft: it resolves, and it says nothing for the field the index declares. While
|
|
// deposits were re-read for ever this was a benign WAIT — the object might gain the field
|
|
// and a later drain would index it — so it was traced behind `debugAccessLog` and nothing
|
|
// more. Consumption ends the wait: this deposit is gone once it has been looked at, so the
|
|
// discard has to be audible without any flag turned on.
|
|
setCurrentUser("alice");
|
|
const draft = await createEntityDoc("alice", "public");
|
|
// It has to SAY something, or the read comes back empty and this is the `refused` path
|
|
// instead — a different outcome with a different report. What is under test is the
|
|
// object that resolves fine and simply has no value for the indexed field.
|
|
await write(draft, "urn:test:title", "a draft with no date yet");
|
|
setCurrentUser(null);
|
|
await bobDepositsInto(index, draft, 1);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /was discarded/.test(l) && new RegExp(`carries no ${WHEN}`).test(l)))
|
|
.toBe(true);
|
|
expect(await entryValuesFor(index, draft)).toEqual([]);
|
|
});
|
|
|
|
// --- a broker that goes quiet must not walk the whole queue -----------------
|
|
|
|
test("a broker that cannot read the referenced objects costs ONE deposit, not the queue", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// Three references waiting in the index's inbox.
|
|
setCurrentUser("alice");
|
|
const objects: Nuri[] = [];
|
|
for (let i = 0; i < 3; i++) {
|
|
const o = await createEntityDoc("alice", "public");
|
|
await write(o, WHEN, `2026-01-0${i + 1}`);
|
|
objects.push(o);
|
|
}
|
|
setCurrentUser(null);
|
|
for (let i = 0; i < 3; i++) await bobDepositsInto(index, objects[i]!, i + 1);
|
|
|
|
const queued = () =>
|
|
fake.quads.filter((q) => q.p === `${INBOX}:payload` && q.o.includes("did:ng:o:")).length;
|
|
expect(queued()).toBe(3);
|
|
|
|
// The broker goes quiet on the referenced objects. The read THROWS, so the arm is told a
|
|
// failure happened rather than left to guess it from an empty answer — a distinction the
|
|
// batch read of the day made with an `onUnreadable` reporter, because it swallowed the
|
|
// failure into `[]`, and `1.0.0-dev.8` makes with the language. Until 2026-08-26 this was
|
|
// classed as an ordinary refusal: the drain reported it and carried on, popping and destroying all three
|
|
// while printing "will not be retried" for each.
|
|
const realQuery = fake.sparql_query;
|
|
fake.sparql_query = (async (...a: unknown[]) => {
|
|
if (objects.includes(a[3] as Nuri)) throw new Error("the broker went quiet");
|
|
return realQuery(...a);
|
|
}) as typeof fake.sparql_query;
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
|
|
expect(queued()).toBe(2);
|
|
});
|
|
|
|
test("an EMPTY but readable object is absent — reported, and the queue behind it carries on", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
// A document Alice really holds and that really reads — it just has no triples. This is
|
|
// the genuine clean-empty read, and until 2026-08-26 NOTHING exercised it: the only test
|
|
// of this branch pointed at `did:ng:o:nowhere`, which the possession gate refuses before
|
|
// any rows exist. So the branch that decides "absent vs unreadable" on a real read was
|
|
// uncovered, which is how two opposite misclassifications shipped through it.
|
|
setCurrentUser("alice");
|
|
const empty = await createEntityDoc("alice", "public");
|
|
setCurrentUser(null);
|
|
|
|
// The empty one FIRST, so "does it stall what is behind it" is a question with an answer.
|
|
await bobDepositsInto(index, empty, 1);
|
|
await bobDepositsInto(index, object, 2);
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /read cleanly from a synced repo and found nothing/.test(l)))
|
|
.toBe(true);
|
|
expect(await entryValuesFor(index, empty)).toEqual([]);
|
|
expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
|
|
});
|
|
|
|
test("a rejection valued `null` is still a failure, not an absence", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
await bobDepositsInto(index, object, 1);
|
|
|
|
// `throw null` is legal, and the first version of this classification tracked the error
|
|
// VALUE against a `null` sentinel — so a broker that rejected with `null` was recorded as
|
|
// "no failure happened" and its deposit was destroyed as an absence. Tracked by a boolean
|
|
// now, which is the only thing that cannot collide with a payload.
|
|
const realQuery = fake.sparql_query;
|
|
fake.sparql_query = (async (...a: unknown[]) => {
|
|
if (a[3] === object) throw null;
|
|
return realQuery(...a);
|
|
}) as typeof fake.sparql_query;
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
|
|
expect(reported.some((l) => /came back clean and empty/.test(l))).toBe(false);
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
});
|
|
|
|
test("an empty read whose repo is NOT SYNCED is not an absence — it stops the drain", async () => {
|
|
inject();
|
|
await bobSignsInOnce();
|
|
const { index, object } = await aliceOpensAnIndex([WHEN]);
|
|
|
|
setCurrentUser("alice");
|
|
const unsynced = await createEntityDoc("alice", "public");
|
|
setCurrentUser(null);
|
|
await bobDepositsInto(index, unsynced, 1);
|
|
await bobDepositsInto(index, object, 2);
|
|
|
|
// A broker that ACCEPTS the subscription and never pushes `State`. The bounded fallback
|
|
// fires, the repo is `timed-out` rather than `synced`, and the anchored read then answers
|
|
// 0 rows WITH NO ERROR — which is the shape a cold start takes, not an exotic one:
|
|
// `Verifier::load` repopulates `self.repos` from storage, so the repo is present and its
|
|
// commits simply have not arrived. Reading that as "the object is absent" consumed and
|
|
// destroyed the whole queue while reporting each deposit as absent. `open-repo` keeps
|
|
// `synced` and `timed-out` apart precisely so this arm can tell, and now it asks.
|
|
setOpenTimeoutForTests(20);
|
|
fake.doc_subscribe = () => () => {};
|
|
|
|
const reported = await reportedDuring(aliceConnects);
|
|
|
|
expect(reported.some((l) => /is not synced in this session/.test(l))).toBe(true);
|
|
expect(reported.some((l) => /came back clean and empty/.test(l))).toBe(false);
|
|
// Stopped rather than carried on: the deposit behind it is untouched.
|
|
expect(await entryValuesFor(index, object)).toEqual([]);
|
|
});
|