/** * An inbox is a QUEUE, and this suite is about the act that had been missing from it: a * deposit is delivered once and then it is gone. * * ── What each test would have looked like before ────────────────────────── * Until 2026-08-26 this package never consumed anything. Every drain re-read everything * ever deposited, which produced three symptoms at once: an inbox that grew without bound, * a malformed deposit re-reported at every push and every connection for as long as the * page lived, and an idempotence burden on every arm that existed only to survive the * replay. Each test below fails against that code — the deposit is still there — and fails * again if the removal is taken back out. * * ── Upstream is the specification, and so is its ORDER ──────────────────── * The broker POPS: `take_first_msg` → `MSGS.take_first_value` is `get_first_key_value` * followed by `del`, so the message leaves the store BEFORE it is delivered; the verifier * then applies it and a failure is logged and swallowed, never re-enqueued * (`engine/verifier/src/verifier.rs`). Remove, apply, log, move on. So a deposit whose arm * refused is gone exactly like one that succeeded, and that is the point of the second test * below rather than an accident of it. * * ── Nobody is handed an inbox address ───────────────────────────────────── * A depositor names a document and a person (`inbox.share`, `inbox.postToDocument`); the * address is resolved inside. The addresses the TEST resolves are read off the wallet and * used only to count what is left in a queue — never given to an actor. */ import { test, expect, describe, afterAll, beforeEach } from "bun:test"; import { inbox as inboxSurface, storeRegistry } from "../src/index"; import { resolveAccount, userInbox } from "../src/shared-wallet/account-registry"; import { setCurrentUser } from "../src/shared-wallet/bootstrap"; import { openDocumentInbox } from "../src/emulated-verifier/branch-registers"; import { cancelScheduledInboxProcessing, runScheduledInboxProcessingNow, } from "../src/emulated-verifier/inbox-processor"; import { bootPage, forgetEverything, signIn, type FakeWallet, type Quad } from "./wallet-fake"; import type { Nuri } from "../src/model/types"; const INBOX = "urn:ng-eventually:inbox"; /** A kind in this package's OWN namespace that no arm answers — upstream's `NotImplemented`. */ const UNARMED_KIND = `${INBOX}:nonesuch`; let quads: Quad[]; let fake: FakeWallet; beforeEach(() => { forgetEverything(); quads = []; fake = bootPage(quads); }); afterAll(() => { cancelScheduledInboxProcessing(); forgetEverything(); }); /** What is STILL QUEUED in `inbox` — one entry per deposit that has not been consumed. */ function stillQueuedIn(inbox: Nuri): string[] { return quads.filter((q) => q.g === inbox && q.p === `${INBOX}:payload`).map((q) => q.o); } /** How many deposits have been RETIRED — one `DELETE WHERE { ?p ?o }` each. */ function removalCount(): number { return fake.sparql_update.mock.calls.filter((c) => /DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(c[1])) ).length; } /** Capture what `console.error` is told while `body` runs — where a refusal goes. */ async function reportedDuring(body: () => Promise): Promise { 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 body(); } finally { console.error = real; } return lines; } /** * Alice has been here once, so she is someone a deposit can NAME (`inbox.share` refuses a * recipient nobody has ever been, and connecting does not provision — an identity acquires * its account the first time it creates anything). */ async function aliceHasBeenHere(): Promise { await signIn("alice"); await storeRegistry.createEntityDoc(await storeRegistry.protectedStore()); } /** Bob shares a document of his with Alice, by name. Answers the document. */ async function bobSharesWithAlice(): Promise { await signIn("bob"); const doc = await storeRegistry.createEntityDoc(await storeRegistry.protectedStore()); await inboxSurface.share(doc, "alice"); return doc; } /** The Links durably filed on a user's User branch — the emulated `AddLink` records. */ async function documentsGivenTo(id: string): Promise { const store = (await resolveAccount(id))?.docPrivate; if (!store) return []; return quads .filter((q) => q.g === store && q.p === "urn:ng-eventually:shim:link") .map((q) => q.o.split(":r:")[0]!); } // --- a deposit that was applied is GONE ----------------------------------- test("a Link that has been applied is no longer in the queue", async () => { await aliceHasBeenHere(); const doc = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); // Bob's deposit is waiting, and nothing of it has reached Alice yet. expect(stillQueuedIn(aliceInbox).length).toBe(1); expect(await documentsGivenTo("alice")).toEqual([]); // Alice connects: her own session drains her own inbox. await signIn("alice"); // Applied — and CONSUMED. The second assertion is the one that was missing: before // 2026-08-26 the Link was filed and the deposit stayed, so this queue only ever grew. expect(await documentsGivenTo("alice")).toEqual([doc]); expect(stillQueuedIn(aliceInbox)).toEqual([]); }); // --- a deposit whose ARM FAILED is gone too, and said so once -------------- describe("a deposit no arm can apply", () => { /** * Alice opens a document with an inbox on it; Bob deposits something that declares * itself to this package and that no arm answers. He names her DOCUMENT — the address is * resolved inside, and he never learns it. */ async function bobDepositsSomethingUnarmed(): Promise { await signIn("alice"); // PUBLIC, so Bob can read it — the address rides on the document, so a depositor who // cannot read it cannot reach its owner. That is the model, not a fixture detail. const note = await storeRegistry.createEntityDoc(await storeRegistry.publicStore()); await openDocumentInbox(note); await signIn("bob"); await inboxSurface.postToDocument(note, { payload: { kind: UNARMED_KIND } }); return note; } test("is reported AND consumed — upstream logs the failure and the message is already gone", async () => { const note = await bobDepositsSomethingUnarmed(); const noteInbox = await (async () => { const rec = quads.find( (q) => q.p === "urn:ng-eventually:shim:inboxCap" && q.o.startsWith(note + " "), ); if (!rec) throw new Error("no AddInboxCap record was written for the note"); return rec.o.split(" ")[1] as Nuri; })(); expect(stillQueuedIn(noteInbox).length).toBe(1); const reported = await reportedDuring(async () => { await signIn("alice"); }); // It said so — a message that could not be applied is never silent. expect(reported.some((l) => /no arm applies deposits of kind/.test(l))).toBe(true); // …and it is GONE. This is the whole change: upstream the broker had already removed // the message before the arm ran, so `log_err!` is the END of it, not a promise to // try again. The old wording ("it stays in it and the next attempt will try again") // described a deposit that would be re-reported forever. expect(stillQueuedIn(noteInbox)).toEqual([]); expect(reported.some((l) => /will not be retried/.test(l))).toBe(true); }); test("is not reported a SECOND time, because a second drain finds an empty queue", async () => { const note = await bobDepositsSomethingUnarmed(); await signIn("alice"); // Alice connects again — a reload, a second tab, any of the things that drain. Before // consumption this reported the same unapplicable deposit again, and would have gone // on doing so at every connection for as long as it sat there. const reportedAgain = await reportedDuring(async () => { await signIn("alice"); }); expect(reportedAgain.filter((l) => /no arm applies deposits of kind/.test(l))).toEqual([]); void note; }); }); // --- a second drain of a drained inbox does nothing ------------------------ test("draining an inbox twice applies once and reports nothing the second time", async () => { await aliceHasBeenHere(); const doc = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); await signIn("alice"); expect(await documentsGivenTo("alice")).toEqual([doc]); const reported = await reportedDuring(async () => { await signIn("alice"); }); // Nothing to do, nothing said, and nothing filed twice. Note WHY that last one is not // just `addLink`'s de-duplication doing the work: the queue is empty, so no arm ran. expect(reported).toEqual([]); expect(stillQueuedIn(aliceInbox)).toEqual([]); expect(await documentsGivenTo("alice")).toEqual([doc]); }); // --- ONE primitive, exercised through BOTH triggers ----------------------- /** * The two triggers must consume through the SAME act, so this asserts on that act rather * than on each trigger — a fault injected at the ONE removal, and the same consequence * demanded of both paths. * * The consequence is `inbox-consume`'s own contract: a removal that fails STOPS the apply, * because applying a deposit that stayed in the queue is the replay this exists to end. A * trigger carrying its own copy of the removal would not be bound by it. */ describe("both triggers consume through the one primitive", () => { /** Make retiring a deposit fail — the single `DELETE WHERE { ?p ?o }`. */ function breakTheRemoval(): void { const real = fake.sparql_update; fake.sparql_update = (async (...a: unknown[]) => { if (/DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(a[1]))) { throw new Error("the broker refused to retire this deposit"); } return real(...a); }) as typeof fake.sparql_update; } test("the OWNER's own drain applies nothing when the deposit cannot be retired", async () => { await aliceHasBeenHere(); await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); breakTheRemoval(); await reportedDuring(async () => { await signIn("alice"); }); // Nothing applied, and the deposit is still queued — so the next drain will try it // again, which is the only safe answer when the pop did not happen. expect(await documentsGivenTo("alice")).toEqual([]); expect(stillQueuedIn(aliceInbox).length).toBe(1); }); test("the DEFERRED stand-in applies nothing either, by the same rule", async () => { await aliceHasBeenHere(); // Alice is NOT connected: Bob's session stands in for her absent one. await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); breakTheRemoval(); await reportedDuring(async () => { await runScheduledInboxProcessingNow(); }); expect(await documentsGivenTo("alice")).toEqual([]); expect(stillQueuedIn(aliceInbox).length).toBe(1); }); test("and with the removal working, the DEFERRED stand-in consumes exactly as the owner does", async () => { await aliceHasBeenHere(); const doc = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); await runScheduledInboxProcessingNow(); // Same two facts the owner's own drain produces above — one act, two triggers. expect(await documentsGivenTo("alice")).toEqual([doc]); expect(stillQueuedIn(aliceInbox)).toEqual([]); }); }); // --- a pop whose effect did not land must SAY so --------------------------- test("an I/O failure while filing a Link STOPS the drain, so the rest of the queue survives", async () => { await aliceHasBeenHere(); // Three deposits waiting for an absent Alice, so "what happened to the OTHERS" is a // question this can answer at all. await bobSharesWithAlice(); await bobSharesWithAlice(); await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); expect(stillQueuedIn(aliceInbox).length).toBe(3); // The `AddLink` write fails — a broker that will not take the commit. const real = fake.sparql_update; fake.sparql_update = (async (...a: unknown[]) => { if (/urn:ng-eventually:shim:link/.test(String(a[1]))) { throw new Error("the broker refused the AddLink commit"); } return real(...a); }) as typeof fake.sparql_update; const reported = await reportedDuring(async () => { await runScheduledInboxProcessingNow(); }); // Reported ungated, at the granularity of the INBOX — a genuine I/O failure is not an // unapplicable item, and the drain stops rather than working through the queue. expect(reported.some((l) => /could not finish draining this inbox/.test(l))).toBe(true); expect(await documentsGivenTo("alice")).toEqual([]); // The load-bearing assertion. One deposit was popped before the write failed and is lost; // the other TWO are untouched and wait for the next attempt. Between 2026-08-26's first // and second corrections this read 0: `applyLink` caught the throw and returned `false`, // so the loop went on popping — and most of what `addLink` does before its write is // deposit-independent infrastructure that fails for the whole queue at once. Catching it // per deposit therefore destroyed all three while dutifully reporting each one. expect(stillQueuedIn(aliceInbox).length).toBe(2); // And there is no WEDGE. Stopping the drain could have traded destruction for permanent // blockage — a deposit that always throws would sit at the head of the queue for ever and // no later one would be reached. It cannot, and the ORDER is why: the pop happens BEFORE // the arm, so the deposit that failed is already gone and the next attempt starts at the // one behind it. A failing broker costs one deposit per attempt, not the queue. // // The attempts are ALICE CONNECTING, not the timer: a deferred window is armed by a // deposit and is not re-armed by a drain that failed, so the owner's own next connection // is the recovery path — which is what the emulated stand-in has always documented. await signIn("alice"); expect(stillQueuedIn(aliceInbox).length).toBe(1); await signIn("alice"); expect(stillQueuedIn(aliceInbox).length).toBe(0); }); // --- delivered ONCE means once, across both triggers ---------------------- test("a deposit the deferred drain consumed is not re-applied by the owner's own next pass", async () => { await aliceHasBeenHere(); await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); // Alice READS her inbox without processing it — which is what `inbox.watch` does on every // push, and it is the only step that records what this session decoded. She is set as the // holder WITHOUT connecting, deliberately: connecting drains, and a drain would consume // the deposit before the read could see it. setCurrentUser("alice"); expect((await inboxSurface.read(aliceInbox)).length + stillQueuedIn(aliceInbox).length) .toBeGreaterThan(0); setCurrentUser("bob"); // The deferred stand-in then drains the same inbox through its own path, which never // touches that record. Before 2026-08-26 the record was only ever WRITTEN when something // was declared, so it survived the queue it described: Alice's own next pass worked from // a list of deposits that were already gone, re-popping and re-applying every one. await runScheduledInboxProcessingNow(); const removalsBefore = removalCount(); await signIn("alice"); expect(removalCount()).toBe(removalsBefore); expect(stillQueuedIn(aliceInbox)).toEqual([]); }); // --- the gap between deciding to deliver and delivering -------------------- test("an identity that moves DURING the pop does not get the deposit applied for it", async () => { await aliceHasBeenHere(); // Alice opens an index and Bob deposits a bare reference into it. That arm is the one // that reads AMBIENT state, so it is the one an identity switch can mis-file. await signIn("alice"); const index = await storeRegistry.createEntityDoc(await storeRegistry.publicStore()); await openDocumentInbox(index); const object = await storeRegistry.createEntityDoc(await storeRegistry.publicStore()); await signIn("bob"); await inboxSurface.postToDocument(index, { payload: object }); const indexInbox = (() => { const rec = quads.find( (q) => q.p === "urn:ng-eventually:shim:inboxCap" && q.o.startsWith(index + " "), ); if (!rec) throw new Error("no AddInboxCap record was written for the index"); return rec.o.split(" ")[1] as Nuri; })(); // Alice signs in, so the reference is CLAIMED — her own session may apply it. Then the // identity moves in the one place nothing used to look: inside the await that removes the // deposit. `claimsDeposit` had already said yes, and the arm below reads whoever holds the // page now, so before the re-check this filed Alice's deposit against Bob's registers. const real = fake.sparql_update; fake.sparql_update = (async (...a: unknown[]) => { const out = await real(...a); if (/DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(a[1]))) { setCurrentUser("bob"); } return out; }) as typeof fake.sparql_update; const reported = await reportedDuring(async () => { await signIn("alice"); }); // The deposit is gone — it was popped before the switch was visible — and that is said // out loud rather than papered over. What must NOT have happened is the arm running for // the wrong holder. expect(stillQueuedIn(indexInbox)).toEqual([]); expect(reported.some((l) => /consumed but not applied/.test(l))).toBe(true); }); test("an arm that fails WHILE the page switches user is still reported, not swallowed", async () => { await aliceHasBeenHere(); await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); // Both at once: the `AddLink` write fails AND the identity moves. `connect.connectedUser` // drops an error from a drain whose holder has gone — an ownership refusal there is // abandoning, not a failure — and that guard cannot tell the two apart. So for one day // this exact pair destroyed a deposit and printed nothing at all. const real = fake.sparql_update; fake.sparql_update = (async (...a: unknown[]) => { if (/urn:ng-eventually:shim:link/.test(String(a[1]))) { setCurrentUser("bob"); throw new Error("the broker refused the AddLink commit"); } return real(...a); }) as typeof fake.sparql_update; const reported = await reportedDuring(async () => { await signIn("alice"); }); // The deposit is gone, and SOMETHING said so. Reported by the only code that knows a // deposit was consumed, before the error is handed to anyone who might drop it. expect(stillQueuedIn(aliceInbox)).toEqual([]); expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true); }); test("a deposit that cannot be retired is refused, not applied again at every drain", async () => { await aliceHasBeenHere(); const doc = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); // A broker that answers the deposits query WITHOUT binding the deposit's subject. `?d` is // bound by the query so this should not happen, but `depositsFrom` defaults it to "" and // an empty subject would issue `DELETE WHERE { <> ?p ?o }`. The load-bearing half is that // it must not be APPLIED either: a deposit applied but never retired is the replay this // whole change exists to end, repeated at every drain for as long as the row sits there. const realQuery = fake.sparql_query; fake.sparql_query = (async (...a: unknown[]) => { const res = await realQuery(...a); if (a[3] === aliceInbox && String(a[1]).includes(`${INBOX}:payload`)) { const r = res as { results?: { bindings?: Array> } }; for (const row of r.results?.bindings ?? []) delete row.d; } return res; }) as typeof fake.sparql_query; const reported = await reportedDuring(async () => { await signIn("alice"); }); expect(reported.some((l) => /carries no subject/.test(l))).toBe(true); expect(await documentsGivenTo("alice")).toEqual([]); expect(stillQueuedIn(aliceInbox).length).toBe(1); void doc; });