diff --git a/docs/bridge.md b/docs/bridge.md index da23705..e899b6c 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -221,7 +221,7 @@ Proposed runtime half (follow-up `persona-wakes-on-address-only`): ### Room tail (`chat_pending.v3`, field 2) -The v3 answer is `[persona, deliveries, tail]`. `tail` is a list of recent room lines, oldest first, for a reader that was not woken for them (a scheduled beat). It uses the same room read as a delivery's `context` (the same function, a separate read), as her account, with no new endpoint; the cost is one `/messages` call per speakable room per poll (15 seconds cached), including every DM. A `chat_pending_request.v1` is still answered with the two-field `chat_pending.v2`. `chat_pending.v3` has not been deployed, but a reader that decodes it as a two-field tuple fails on the third element (rmp-serde `LengthMismatch`), so Huginn must accept `tail` (a defaulted field) before this mouth is deployed. +The v3 answer is `[persona, deliveries, tail, tail_gaps]`. `tail` is a list of recent room lines, oldest first, for a reader that was not woken for them (a scheduled beat). It uses the same room read as a delivery's `context` (the same function, a separate read), as her account, with no new endpoint; the cost is one `/messages` call per speakable room per poll (15 seconds cached), including every DM. A `chat_pending_request.v1` is still answered with the two-field `chat_pending.v2`. `chat_pending.v3` has not been deployed, but a reader that decodes it as a two-field tuple fails on the third element (rmp-serde `LengthMismatch`), so Huginn must accept `tail` and `tail_gaps` (defaulted fields) before this mouth is deployed. The array is positional: a decoder of the earlier three-field shape fails on the fourth element with the same `LengthMismatch`, so `tail_gaps` has the same deploy precondition; a decoder that reads the first three elements and ignores the rest sees `persona`, `deliveries` and `tail` unchanged. | Bound | Rule | |---|---| @@ -234,7 +234,24 @@ A line is the 7-field tuple `[room, event_id, at_unix_ms, sender, form, basis, b The tail is **data, not instruction**. A line is what a sender typed in a room, and the mouth wakes no one, acks nothing and sends nothing because of it. A reader treats the body as untrusted text and weighs `basis` as evidence, as for a delivery. The tail does not replace a delivery: a message still reaches her as a delivery, with its own `seq`, and only `chat_ack` advances a position. -Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries two lines); Huginn pins the same bytes. +#### Gaps in the tail (`chat_pending.v3`, field 3, `tail_gaps`) + +`tail_gaps` is a list of room names (the configured alias, or the room id for a DM, as `room` in a tail line), one for each speakable room whose lines are not all in `tail`. A reader renders such a room as **unknown**, not quiet. Absent or empty means every speakable room was read fully: a room with no lines because it is quiet, and a room whose few lines all fit, are never listed. Order is the order the mouth reads rooms (configured rooms in join order, then DMs), not time; a room appears once however many causes apply. Causes, all of them: + +| Cause | Where | +|---|---| +| The room's read failed (the homeserver refused or errored) or did not answer within 10 seconds (`sendTimeoutMs`); the room is logged and contributes no lines | the mouth | +| The room had more than 20 lines outside threads (the per-room cap; her own lines count), so older ones were left out | the mouth | +| The read window came back full (100 raw events, before edits and thread replies are set aside), so real lines older than the window may exist and are unseen; a shorter chunk means the room's history was exhausted. A room whose history is exactly 100 events long is gapped too, since the mouth cannot tell. The rule assumes a homeserver returns the 100 events asked for whenever older history exists; a server that caps the limit lower turns a truncated read into a false "whole", and Bifrost does not detect that | the mouth | +| One of the room's lines was cut by the 60-line total cap (only the newest 60 over all rooms are kept) | the mouth | +| A line was dropped as malformed (no event id) | the mouth | +| The reply byte budget (256 KiB, deliveries first) cut the room's oldest lines, or a line would not encode and was skipped | the reply | +| The mouth has not started: every configured room is a gap (DMs are not yet known) | the mouth | +| The whole tail read failed: every speakable room is a gap, and the deliveries are still served | the endpoint | + +A line a delivery already shows and a room she cannot speak in change nothing here; her own lines are lines like any other and count toward the caps. A read the mouth kept for 15 seconds counts as a read. + +Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries two lines and no gap, `chat_pending.v3.tail-gaps` one line and two gap rooms; the earlier v3 samples gain an empty fourth element); Huginn pins the same bytes. Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority. diff --git a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack index 6685251..f72544e 100644 Binary files a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack and b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack differ diff --git a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail-gaps.msgpack b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail-gaps.msgpack new file mode 100644 index 0000000..e622eb7 Binary files /dev/null and b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail-gaps.msgpack differ diff --git a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail.msgpack b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail.msgpack index 0ab905b..32e6676 100644 Binary files a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail.msgpack and b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail.msgpack differ diff --git a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.unproven-notice.msgpack b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.unproven-notice.msgpack index 675fdd3..987ac88 100644 Binary files a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.unproven-notice.msgpack and b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.unproven-notice.msgpack differ diff --git a/tests/persona-matrix-mouth-runtime.test.mjs b/tests/persona-matrix-mouth-runtime.test.mjs index e2c9273..b9361cc 100644 --- a/tests/persona-matrix-mouth-runtime.test.mjs +++ b/tests/persona-matrix-mouth-runtime.test.mjs @@ -94,7 +94,7 @@ test("the brake reader reads a real CultCache control store", async () => { test("two clients served at once each get their own answer on their own session", async () => { const mouth = { pendingAfter: () => [], - tail: async () => [], + tail: async () => ({ lines: [], gaps: [], rooms: [] }), ack: async () => {}, // Slow enough that the two requests are in flight together. send: async request => { await new Promise(r => setTimeout(r, 150)); return { requestId: request.requestId, eventId: `$${request.requestId}`, refusal: null }; }, @@ -140,7 +140,8 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a const calls = { acks: [], sends: [], reads: [] }; const mouth = { pendingAfter: after => (after < 7 ? [delivery] : []), - tail: async () => { if (calls.tailFails) throw new Error("room read failed"); return [tailLine]; }, + tail: async () => { if (calls.tailFails) throw new Error("room read failed"); return { lines: [tailLine], gaps: ["#operator:gamecult.org"], rooms: ["#operator:gamecult.org", "#swarm:gamecult.org"] }; }, + speakableRooms: () => ["#operator:gamecult.org", "#swarm:gamecult.org"], ack: async through => { calls.acks.push(through); }, read: async request => { calls.reads.push(request); return null; }, send: async request => { calls.sends.push(request); return request.text === "muted" ? { requestId: request.requestId, eventId: null, refusal: "Brake" } : { requestId: request.requestId, eventId: "$sent", refusal: null }; }, @@ -175,6 +176,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a const pending = docs.parsePending(msgpack.decode(document.payload)); assert.deepEqual(pending.deliveries.map(d => d.eventId), ["$e1"]); assert.deepEqual(pending.deliveries[0], delivery, "sender, mentions and the DM flag cross the wire exactly"); + assert.deepEqual(Buffer.from(document.payload), await readFile(resolve(root, "tests", "fixtures", "persona-chat-v3", "gamecult.persona.chat_pending.v2.no-image.msgpack")), "a v1 request is answered with the two-field v2, byte for byte the golden, whatever the tail gaps"); reply = nextReply(); put(docs.kinds.pendingRequestV3, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 })); @@ -183,6 +185,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a const v3 = docs.parsePendingV3(msgpack.decode(document.payload)); assert.deepEqual(v3.deliveries[0], { ...delivery, basis: null, form: "message" }); assert.deepEqual(v3.tail, [tailLine], "the v3 answer carries the room tail"); + assert.deepEqual(v3.tailGaps, ["#operator:gamecult.org"], "and the rooms whose tail is incomplete"); calls.tailFails = true; reply = nextReply(); @@ -191,6 +194,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a const degraded = docs.parsePendingV3(msgpack.decode(document.payload)); assert.equal(degraded.deliveries.length, 1, "a tail that throws does not silence the deliveries"); assert.deepEqual(degraded.tail, []); + assert.deepEqual(degraded.tailGaps, ["#operator:gamecult.org", "#swarm:gamecult.org"], "a tail that throws leaves every speakable room a gap, not quiet"); calls.tailFails = false; put(docs.kinds.ack, "eureka:7", docs.ackTuple({ persona: "eureka", throughSeq: 7 })); @@ -340,6 +344,9 @@ const v5Samples = { { roomAlias: "#operator:gamecult.org", eventId: "$e0", atUnixMs: 1_700_000_000_000, sender: "@meta:gamecult.org", form: "message", basis: null, text: "earlier" }, { roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 1_700_000_001_000, sender: "@meta:gamecult.org", form: "notice", basis: "tag", text: "hello eureka" }, ] }), parse: docs.parsePendingV3 }, + "gamecult.persona.chat_pending.v3.tail-gaps": { tuple: docs.pendingV3Tuple({ persona: "eureka", deliveries: [], tail: [ + { roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 1_700_000_001_000, sender: "@meta:gamecult.org", form: "message", basis: null, text: "hello eureka" }, + ], tailGaps: ["#swarm:gamecult.org", "!dm1:gamecult.org"] }), parse: docs.parsePendingV3 }, "gamecult.persona.chat_pending.v3.unproven-notice": { tuple: docs.pendingV3Tuple({ persona: "eureka", deliveries: [{ ...delivery, isDirect: false, mentions: [], basis: null, form: "notice" }] }), parse: docs.parsePendingV3 }, }; for (const [name, { tuple, parse }] of Object.entries(v5Samples)) { @@ -349,9 +356,10 @@ for (const [name, { tuple, parse }] of Object.entries(v5Samples)) { if (process.env.UPDATE_FIXTURES === "1") { await mkdir(v5Dir, { recursive: true }); await writeFile(file, bytes); } assert.deepEqual(bytes, await readFile(file), "wire bytes changed: Huginn decodes these fixtures"); assert.deepEqual(parse(msgpack.decode(bytes)), parse(tuple)); - if (name.includes("pending.v3")) assert.equal(msgpack.decode(bytes).length, 3, "persona, deliveries, tail"); + if (name.includes("pending.v3")) assert.equal(msgpack.decode(bytes).length, 4, "persona, deliveries, tail, tail_gaps"); if (name.endsWith("v3.tag") || name.endsWith("unproven-notice")) assert.equal(msgpack.decode(bytes)[1][0].length, 14); - if (name.endsWith("v3.tail")) assert.equal(msgpack.decode(bytes)[2][0].length, 7); + if (name.endsWith("v3.tail")) { assert.equal(msgpack.decode(bytes)[2][0].length, 7); assert.deepEqual(msgpack.decode(bytes)[3], [], "no gap: every room read fully"); } + if (name.endsWith("v3.tail-gaps")) assert.deepEqual(msgpack.decode(bytes)[3], ["#swarm:gamecult.org", "!dm1:gamecult.org"]); }); } @@ -390,7 +398,7 @@ test("RoomMoved is v2 only, and the v2 documents are strict about arity and newe test("the RUDP endpoint answers a v1 send with the three-field receipt and a v2 send with the four-field one", async () => { const seen = []; - const mouth = { pendingAfter: () => [], tail: async () => [], ack: async () => {}, send: async request => { + const mouth = { pendingAfter: () => [], tail: async () => ({ lines: [], gaps: [], rooms: [] }), ack: async () => {}, send: async request => { seen.push(request); return request.basis ? { requestId: request.requestId, eventId: null, refusal: "RoomMoved", newer: ["$a"] } : { requestId: request.requestId, eventId: "$sent", refusal: null, newer: [] }; } }; @@ -566,3 +574,16 @@ test("the RUDP endpoint answers an open-direct request with a receipt for its Pe try { client.close(); } catch { /* the connection already closed its socket */ } } }); + +test("tail_gaps is additive: a three-field reader finds persona, deliveries and tail unchanged in the first three elements of the new bytes", async () => { + const sample = { persona: "eureka", deliveries: [{ ...delivery, isDirect: false, basis: "tag", form: "message" }], tail: [{ roomAlias: "#operator:gamecult.org", eventId: "$e0", atUnixMs: 1_700_000_000_000, sender: "@meta:gamecult.org", form: "message", basis: null, text: "earlier" }] }; + const noGaps = msgpack.decode(msgpack.encode(docs.pendingV3Tuple(sample))); + const after = msgpack.decode(await readFile(resolve(root, "tests", "fixtures", "persona-chat-v5", "gamecult.persona.chat_pending.v3.tail-gaps.msgpack"))); + const withGaps = msgpack.decode(msgpack.encode(docs.pendingV3Tuple({ ...sample, tailGaps: ["#swarm:gamecult.org"] }))); + assert.equal(withGaps.length, 4); + assert.deepEqual(withGaps.slice(0, 3), noGaps.slice(0, 3), "persona, deliveries and tail are byte-for-meaning the same with or without gaps"); + assert.deepEqual(docs.parsePendingV3([...withGaps.slice(0, 3), []]).tail, docs.parsePendingV3(noGaps).tail); + assert.equal(after.length, 4); + // Not claimed: a strict three-field decoder (rmp-serde into a three-field tuple) accepts it. It fails on the length, as it did for the tail, so Huginn defaults tail_gaps before this mouth deploys. + assert.notEqual(withGaps.length, 3); +}); diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs index e4c7731..b2ed814 100644 --- a/tests/persona-matrix-mouth.test.mjs +++ b/tests/persona-matrix-mouth.test.mjs @@ -1485,7 +1485,8 @@ async function tailed({ preload, direct, member, events = [], act = async () => const mouth = mouthFor(hs); await mouth.syncOnce(); await act(mouth); - return { tail: await mouth.tail(), mouth, hs }; + const read = await mouth.tail(); + return { tail: read.lines, gaps: read.gaps, rooms: read.rooms, mouth, hs }; } finally { await hs.close(); } } @@ -1538,10 +1539,10 @@ test("room tail: acking a delivery does not hide it, so an unaddressed line she const mouth = mouthFor(hs); await mouth.syncOnce(); const made = await mouth.syncOnce(); - assert.deepEqual((await mouth.tail()).map(l => [l.eventId, l.basis]), [["$a", null], ["$b", "tag"]]); + assert.deepEqual((await mouth.tail()).lines.map(l => [l.eventId, l.basis]), [["$a", null], ["$b", "tag"]]); await mouth.ack(made[1].seq); assert.equal(mouth.pendingAfter(0).length, 0); - assert.deepEqual((await mouth.tail()).map(l => l.eventId), ["$a", "$b"], "acked lines are the ones no delivery will bring again"); + assert.deepEqual((await mouth.tail()).lines.map(l => l.eventId), ["$a", "$b"], "acked lines are the ones no delivery will bring again"); } finally { await hs.close(); } }); @@ -1551,15 +1552,16 @@ test("room tail: a room that never answers is absent after the timeout and does try { const mouth = mouthFor(hs, { fetch: hang, sendTimeoutMs: 50 }); await mouth.syncOnce(); - const tail = await mouth.tail(); - assert.deepEqual([...new Set(tail.map(l => l.roomAlias))], [`#operator:${HS_NAME}`]); + const read = await mouth.tail(); + assert.deepEqual([...new Set(read.lines.map(l => l.roomAlias))], [`#operator:${HS_NAME}`]); + assert.deepEqual(read.gaps, [`#swarm:${HS_NAME}`], "the room that timed out is a gap; the one that answered is not"); } finally { await hs.close(); } }); test("room tail: a mouth that has not started reads no room", async () => { const hs = await fakeHomeserver({ preload: { [ROOM]: lines(ROOM, 2) }, sync: since => ({ next_batch: since ?? "b0" }) }); try { - assert.deepEqual(await mouthFor(hs).tail(), []); + assert.deepEqual(await mouthFor(hs).tail(), { lines: [], gaps: Object.keys(ROOMS), rooms: Object.keys(ROOMS) }, "no room is known before start, so every configured room is a gap"); assert.equal(hs.seen.filter(r => r.path.includes("/messages")).length, 0); } finally { await hs.close(); } }); @@ -1571,24 +1573,151 @@ test("room tail: a body is clipped like a context line, and an unreadable room i try { const mouth = mouthFor(hs); await mouth.syncOnce(); - assert.deepEqual(await mouth.tail(), []); + const read = await mouth.tail(); + assert.deepEqual(read.lines, []); + assert.deepEqual(read.gaps, Object.keys(ROOMS).filter(a => read.rooms.includes(a)), "an unreadable room is a gap, not quiet"); + assert.ok(read.gaps.length > 0); } finally { await hs.close(); } }); test("room tail: only the v3 answer carries it; the v2 answer a v1 request gets is unchanged", async () => { - const { tail } = await tailed({ preload: { [ROOM]: lines(ROOM, 2) } }); + const { tail, gaps, rooms } = await tailed({ preload: { [ROOM]: lines(ROOM, 2) } }); const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; - assert.deepEqual(docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, tail)).tail.map(l => l.text), ["line 0", "line 1"]); - const v2 = pageOf("eureka", [], pack, () => {}, false, tail); + assert.deepEqual(docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, { lines: tail, gaps, rooms })).tail.map(l => l.text), ["line 0", "line 1"]); + const v2 = pageOf("eureka", [], pack, () => {}, false, { lines: tail, gaps: rooms, rooms }); assert.equal(v2.length, 2); assert.deepEqual(docs.parsePending(v2), { persona: "eureka", deliveries: [] }); - assert.throws(() => docs.parsePendingV3(["eureka", []]), /3-field/); + assert.throws(() => docs.parsePendingV3(["eureka", []]), /4-field/); }); test("room tail: the reply byte budget trims the oldest tail lines, never a delivery", async () => { const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; const big = Array.from({ length: 60 }, (_, i) => ({ roomAlias: "#r:x", eventId: `$${i}`, atUnixMs: i, sender: "@a:x", form: "message", basis: null, text: "y".repeat(5000) })); - const kept = docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, big)).tail; + const kept = docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, { lines: big, gaps: [], rooms: ["#r:x"] })).tail; assert.ok(kept.length > 0 && kept.length < 60); assert.equal(kept.at(-1).eventId, "$59", "the newest survive"); }); + +// tail_gaps (docs/bridge.md, "Gaps in the tail"): the rooms whose lines are not all in the tail, so a reader says "unknown", not "quiet". +// Each case goes through the composition the RUDP endpoint serves: mouth.tail() into pageOf into the strict v3 parser. +const replied = (read, pack = { encode: v => Buffer.from(JSON.stringify(v)) }) => docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, read)); +const SWARM_ALIAS = `#swarm:${HS_NAME}`, OPERATOR_ALIAS = `#operator:${HS_NAME}`; + +test("tail gaps: a room whose read times out is a gap; the room that answered, quiet or not, is not", async () => { + const hs = await fakeHomeserver({ preload: { [OPERATOR]: lines(OPERATOR, 2) }, sync: since => ({ next_batch: since ?? "b0" }) }); + const hang = (url, init) => String(url).includes(encodeURIComponent(ROOM)) && String(url).includes("/messages") ? new Promise(() => {}) : hs.fetch(url, init); + try { + const mouth = mouthFor(hs, { fetch: hang, sendTimeoutMs: 50 }); + await mouth.syncOnce(); + const got = replied(await mouth.tail()); + assert.deepEqual(got.tailGaps, [SWARM_ALIAS]); + assert.deepEqual(got.tail.map(l => l.roomAlias), [OPERATOR_ALIAS, OPERATOR_ALIAS]); + } finally { await hs.close(); } +}); + +test("tail gaps: a failed read is a gap, a quiet room (no lines) and a normal room (a few lines) are not, and exactly the per-room cap is not a trim", async () => { + const failing = await fakeHomeserver({ failContext: true, sync: since => ({ next_batch: since ?? "b0" }) }); + try { + const mouth = mouthFor(failing); + await mouth.syncOnce(); + assert.deepEqual(replied(await mouth.tail()).tailGaps, [SWARM_ALIAS, OPERATOR_ALIAS], "every read failed: both rooms unknown, in room order"); + } finally { await failing.close(); } + + const { gaps } = await tailed({ preload: { [ROOM]: lines(ROOM, docs.limits.tailPerRoom) } }); + assert.deepEqual(gaps, [], "20 lines in one room and none in the other: nothing was left out"); + const some = await tailed({ preload: { [ROOM]: lines(ROOM, 3), [OPERATOR]: [] } }); + assert.deepEqual(replied({ lines: some.tail, gaps: some.gaps, rooms: some.rooms }).tailGaps, []); +}); + +test("tail gaps: a room with more lines than the per-room cap is a gap, once, and only that room", async () => { + const { tail, gaps } = await tailed({ preload: { [ROOM]: lines(ROOM, docs.limits.tailPerRoom + 1), [OPERATOR]: lines(OPERATOR, 3) } }); + assert.equal(tail.filter(l => l.roomAlias === SWARM_ALIAS).length, docs.limits.tailPerRoom); + assert.deepEqual(gaps, [SWARM_ALIAS]); +}); + +test("tail gaps: the total cap names the rooms it dropped lines from, in room order, never a room whose lines all fit", async () => { + const dms = ["!d1:gamecult.test", "!d2:gamecult.test", "!d3:gamecult.test"]; + const preload = { [ROOM]: lines(ROOM, 25, 100), [OPERATOR]: lines(OPERATOR, 20, 0) }; + dms.forEach((dm, i) => { preload[dm] = lines(dm, 20, 200 + 20 * i); }); + const { tail, gaps } = await tailed({ preload, direct: { [META]: dms }, member: [...Object.values(ROOMS), ...dms] }); + assert.equal(tail.length, 60); + assert.deepEqual(gaps, [SWARM_ALIAS, OPERATOR_ALIAS], "swarm is gapped by both caps and named once; the three DMs kept all 20 lines"); +}); + +test("tail gaps: the reply byte budget and a line that will not encode gap their rooms, and only those", () => { + const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; + const mk = (room, i, extra = {}) => ({ roomAlias: room, eventId: `$${room}${i}`, atUnixMs: i, sender: "@a:x", form: "message", basis: null, text: "y".repeat(5000), ...extra }); + const old = Array.from({ length: 30 }, (_, i) => mk("#old:x", i)), fresh = Array.from({ length: 30 }, (_, i) => mk("#new:x", 100 + i)); + const got = replied({ lines: [...old, ...fresh], gaps: [], rooms: ["#old:x", "#new:x", "#empty:x"] }, pack); + assert.ok(got.tail.length < 60); + assert.deepEqual(got.tailGaps, ["#old:x"], "the oldest lines were trimmed; the newer room and the quiet room are whole"); + const bad = replied({ lines: [mk("#a:x", 1), mk("#b:x", 2, { sender: 123 })], gaps: [], rooms: ["#a:x", "#b:x"] }, pack); + assert.deepEqual(bad.tailGaps, ["#b:x"]); +}); + +test("tail gaps: the parser refuses a repeated room, and a pre-gap decoder is stopped by arity, as the tail was", () => { + assert.throws(() => docs.parsePendingV3(["eureka", [], [], ["#a:x", "#a:x"]]), /repeats/); + assert.deepEqual(docs.parsePendingV3(["eureka", [], [], ["#a:x", "#b:x"]]).tailGaps, ["#a:x", "#b:x"]); + assert.throws(() => docs.parsePendingV3(["eureka", [], []]), /4-field/); + assert.throws(() => docs.parsePendingV3(["eureka", [], [], "#a:x"]), /list/); +}); + +test("tail gaps: a line the mouth drops as malformed (no event id) gaps its room; a clean room beside it does not", async () => { + const { tail, gaps } = await tailed({ preload: { [ROOM]: [message("", META, "no id", 10), message("$ok", META, "fine", 11)], [OPERATOR]: lines(OPERATOR, 1) } }); + assert.deepEqual(tail.map(l => l.eventId).sort(), ["$ok", "$ope0"].sort()); + assert.deepEqual(gaps, [SWARM_ALIAS]); +}); + +const edit = (id, ts) => ({ type: "m.room.message", event_id: id, sender: META, origin_server_ts: ts, content: { msgtype: "m.text", body: "* edited", "m.relates_to": { rel_type: "m.replace", event_id: "$x" } } }); +const threaded = (id, ts) => ({ type: "m.room.message", event_id: id, sender: META, origin_server_ts: ts, content: { msgtype: "m.text", body: "in thread", "m.relates_to": { rel_type: "m.thread", event_id: "$root" } } }); +const noise = (n, from) => Array.from({ length: n }, (_, i) => (i % 2 ? threaded : edit)(`$n${from + i}`, 1000 + from + i)); + +test("tail gaps: a full read window hides older real lines behind edits and thread replies, so the room is a gap; a short chunk is exhausted history, not a gap", async () => { + const hidden = await tailed({ preload: { [ROOM]: [...lines(ROOM, 5), ...noise(120, 0)] } }); + assert.deepEqual(hidden.tail, [], "the five real lines lie beyond the 100-event window"); + assert.deepEqual(hidden.gaps, [SWARM_ALIAS], "so the room is unknown, not quiet"); + const whole = await tailed({ preload: { [ROOM]: [...lines(ROOM, 5), ...noise(60, 0)] } }); + assert.equal(whole.tail.length, 5); + assert.deepEqual(whole.gaps, [], "a chunk shorter than the window reached the room's start"); + const at99 = await tailed({ preload: { [ROOM]: [...lines(ROOM, 5), ...noise(94, 0)] } }); + assert.deepEqual(at99.gaps, [], "99 raw events: one short of the window, history exhausted"); + const at100 = await tailed({ preload: { [ROOM]: [...lines(ROOM, 5), ...noise(95, 0)] } }); + assert.deepEqual(at100.gaps, [SWARM_ALIAS], "exactly 100 raw events fill the window, so older history cannot be ruled out"); +}); + +test("tail gaps: the total cap boundary: exactly 60 lines over all rooms is no gap, a 61st names the room that lost its oldest line", async () => { + const dms = ["!d1:gamecult.test", "!d2:gamecult.test"]; + const rooms60 = { [ROOM]: lines(ROOM, 20, 0), [OPERATOR]: lines(OPERATOR, 20, 20), [dms[0]]: lines(dms[0], 20, 40) }; + const at60 = await tailed({ preload: rooms60, direct: { [META]: dms }, member: [...Object.values(ROOMS), ...dms] }); + assert.equal(at60.tail.length, 60); + assert.deepEqual(at60.gaps, []); + const at61 = await tailed({ preload: { ...rooms60, [dms[1]]: lines(dms[1], 1, 60) }, direct: { [META]: dms }, member: [...Object.values(ROOMS), ...dms] }); + assert.equal(at61.tail.length, 60); + assert.deepEqual(at61.gaps, [SWARM_ALIAS]); +}); + +test("tail gaps: a byte-budget cut is contiguous at the oldest end, so an older line that would still fit is cut and its room gapped", () => { + const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; + const mk = (room, i, text) => ({ roomAlias: room, eventId: `$${room}${i}`, atUnixMs: i, sender: "@a:x", form: "message", basis: null, text }); + const tiny = mk("#tiny:x", 0, "t"), big = Array.from({ length: 70 }, (_, i) => mk("#big:x", 1 + i, "y".repeat(5000))); + const got = replied({ lines: [tiny, ...big], gaps: [], rooms: ["#tiny:x", "#big:x"] }, pack); + assert.ok(!got.tail.some(l => l.roomAlias === "#tiny:x"), "nothing older than the cut is kept"); + assert.deepEqual(got.tailGaps, ["#tiny:x", "#big:x"]); +}); + +test("tail gaps: the read timer is cleared when the read answers, and a mouth that has not started names its configured rooms for an endpoint that cannot read", async () => { + const delay = 4321, made = new Set(), cleared = new Set(); + const realSet = globalThis.setTimeout, realClear = globalThis.clearTimeout; + globalThis.setTimeout = (fn, ms, ...rest) => { const t = realSet(fn, ms, ...rest); if (ms === delay) made.add(t); return t; }; + globalThis.clearTimeout = t => { cleared.add(t); return realClear(t); }; + const hs = await fakeHomeserver({ preload: { [ROOM]: lines(ROOM, 2), [OPERATOR]: lines(OPERATOR, 2) }, sync: since => ({ next_batch: since ?? "b0" }) }); + try { + const mouth = mouthFor(hs, { sendTimeoutMs: delay }); + assert.deepEqual(mouth.speakableRooms(), Object.keys(ROOMS), "before start: the configured rooms"); + await mouth.syncOnce(); + assert.deepEqual(mouth.speakableRooms(), Object.keys(ROOMS), "after start: the rooms she can speak in"); + await mouth.tail(); + assert.ok(made.size >= 2); + assert.ok([...made].every(t => cleared.has(t)), "no read timer is left running"); + } finally { globalThis.setTimeout = realSet; globalThis.clearTimeout = realClear; await hs.close(); } +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs index e47b7c8..7dc790d 100644 --- a/tools/persona-chat-documents.mjs +++ b/tools/persona-chat-documents.mjs @@ -247,16 +247,20 @@ function parseTailLine(t) { return { roomAlias: text(roomAlias, "room"), eventId: text(eventId, "event_id"), atUnixMs: seq(atUnixMs, "at_unix_ms"), sender: text(sender, "sender"), form: form(f), basis: basis(b), text: String(body) }; } -// The v3 answer: the persona, the deliveries, then the tail (at most tailPerRoom lines a room, tailTotal in all, oldest first). -export function pendingV3Tuple({ persona, deliveries, tail = [] }) { +// The v3 answer: the persona, the deliveries, the tail (at most tailPerRoom lines a room, tailTotal in all, oldest first), then +// tail_gaps: the rooms (alias, or room id for a DM) whose lines are not all in the tail, each once. Absent or empty: every room was read fully. +export function pendingV3Tuple({ persona, deliveries, tail = [], tailGaps = [] }) { if (tail.length > limits.tailTotal) throw new Error("chat_pending tail is over the line bound"); - return [text(persona, "persona"), deliveries.map(deliveryV3Tuple), tail.map(tailLineTuple)]; + return [text(persona, "persona"), deliveries.map(deliveryV3Tuple), tail.map(tailLineTuple), tailGaps.map(room => text(room, "tail_gap"))]; } export function parsePendingV3(t) { - const [persona, deliveries, tail] = tuple(t, 3, "chat_pending_v3"); + const [persona, deliveries, tail, tailGaps] = tuple(t, 4, "chat_pending_v3"); if (!Array.isArray(deliveries)) throw new Error("chat_pending deliveries must be a list"); if (!Array.isArray(tail)) throw new Error("chat_pending tail must be a list"); - return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDeliveryV3), tail: tail.map(parseTailLine) }; + if (!Array.isArray(tailGaps)) throw new Error("chat_pending tail_gaps must be a list"); + const gaps = tailGaps.map(room => text(room, "tail_gap")); + if (new Set(gaps).size !== gaps.length) throw new Error("chat_pending tail_gaps repeats a room"); + return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDeliveryV3), tail: tail.map(parseTailLine), tailGaps: gaps }; } // The CultNet document_put_raw envelope every chat_* document travels in. diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index d6b4f94..c9767c3 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -254,11 +254,14 @@ export function createMouth({ const lineOf = e => ({ sender: e.sender, text: bodyText(e.content, docs.limits.contextBytes), atUnixMs: Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0 }); // The room's recent messages outside threads, oldest first. - async function roomEvents(roomId) { + // `truncated`: the homeserver filled the whole window, so older history may exist that this read never saw. + async function roomRead(roomId) { const filter = encodeURIComponent(JSON.stringify({ types: ["m.room.message"] })); const reply = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/messages?dir=b&limit=${FETCH_LIMIT}&filter=${filter}`); - return (reply.chunk ?? []).filter(e => isLine(e) && threadRootOf(e) === null).reverse(); + const chunk = reply.chunk ?? []; + return { events: chunk.filter(e => isLine(e) && threadRootOf(e) === null).reverse(), truncated: chunk.length >= FETCH_LIMIT }; } + const roomEvents = async roomId => (await roomRead(roomId)).events; // One thread's messages, root first. async function threadEvents(roomId, rootId) { const room = encodeURIComponent(roomId); @@ -483,33 +486,49 @@ export function createMouth({ // oldest first. From the same /messages read as a delivery's context, as her account, bodies clipped as context lines are. The // basis is what Matrix says on the event itself (no reply lookup: nil is unproven). A room that cannot be read in sendTimeoutMs // is logged and absent, so one stuck room never holds the reply. + // Returns { lines, gaps, rooms }: `rooms` are the rooms read, in read order (configured rooms in join order, then DMs); `gaps` are + // those of them whose lines are not all in `lines` (read failed or timed out, the read window full so older lines may be unseen, a line past a cap, a line dropped as malformed), in + // the same order. A room with nothing to show because it is quiet is not a gap. A mouth that has not started knows no room but the + // configured ones, and all of those are gaps. async function tail() { - if (!started) return []; + if (!started) return { lines: [], gaps: [...rooms], rooms: [...rooms] }; const targets = new Set(); for (const alias of [...joinedAliases, ...directRooms]) { const id = speakableRoomId(alias); if (id) targets.add(id); } + const gapped = new Set(); const perRoom = await Promise.all([...targets].map(async roomId => { + const alias = aliasByRoomId.get(roomId) ?? roomId; let read = tailReads.get(roomId); if (!read || now() - read.at >= TAIL_TTL_MS) { let timer; const lines = Promise.race([ - roomEvents(roomId), + roomRead(roomId), new Promise((_, fail) => { timer = setTimeout(() => fail(new Error(`no answer in ${sendTimeoutMs} ms`)), sendTimeoutMs); }), ]).catch(error => { say(`tail for ${roomId} not fetched: ${error.message}`); return null; }).finally(() => clearTimeout(timer)); read = { at: now(), lines }; tailReads.set(roomId, read); } - const events = await read.lines; - if (events === null) { if (tailReads.get(roomId) === read) tailReads.delete(roomId); return []; } - const alias = aliasByRoomId.get(roomId) ?? roomId; + const got = await read.lines; + if (got === null) { if (tailReads.get(roomId) === read) tailReads.delete(roomId); gapped.add(alias); return []; } + const { events, truncated } = got; + if (truncated || events.length > docs.limits.tailPerRoom) gapped.add(alias); const lines = []; for (const e of events.slice(-docs.limits.tailPerRoom)) { - if (typeof e.event_id !== "string" || e.event_id === "") continue; + if (typeof e.event_id !== "string" || e.event_id === "") { gapped.add(alias); continue; } const at = Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0; lines.push({ roomAlias: alias, eventId: e.event_id, atUnixMs: at, sender: e.sender, form: formOf(e.content), basis: await basisOf(roomId, e, false), text: bodyText(e.content, docs.limits.contextBytes) }); } return lines; })); - return perRoom.flat().sort((a, b) => a.atUnixMs - b.atUnixMs).slice(-docs.limits.tailTotal); + const newest = perRoom.flat().sort((a, b) => a.atUnixMs - b.atUnixMs); + for (const dropped of newest.slice(0, Math.max(0, newest.length - docs.limits.tailTotal))) gapped.add(dropped.roomAlias); + const order = [...targets].map(id => aliasByRoomId.get(id) ?? id); + return { lines: newest.slice(-docs.limits.tailTotal), gaps: order.filter(a => gapped.has(a)), rooms: order }; + } + // The rooms `tail` would read, in its order: what a reply names as unread when the read itself fails outright. + function speakableRooms() { + const ids = new Set(); + for (const alias of [...joinedAliases, ...directRooms]) { const id = speakableRoomId(alias); if (id) ids.add(id); } + return started ? [...ids].map(id => aliasByRoomId.get(id) ?? id) : [...rooms]; } function pendingAfter(afterSeq) { @@ -666,7 +685,7 @@ export function createMouth({ } } - return { start, syncOnce, ack, read, pendingAfter, tail, send, openDirect, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; + return { start, syncOnce, ack, read, pendingAfter, tail, speakableRooms, send, openDirect, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; } // Replace every secret in a message. `secrets` is one string or a list. @@ -711,8 +730,10 @@ export function connectionIdFor(connectionKey) { // The reply to a chat_pending_request: deliveries oldest first, each encoded on its own so one that // will not encode is skipped and logged without silencing the rest, up to the reply byte budget. // The caller acks what it has read and asks again. -// A v3 reply also carries `tail` (room lines, oldest first), trimmed oldest first to what the byte budget has left after the deliveries. -export function pageOf(persona, deliveries, msgpack, log = () => {}, v3 = false, tail = []) { +// A v3 reply also carries `tail` (room lines, oldest first), trimmed oldest first to what the byte budget has left after the deliveries, +// and `tail_gaps`: the rooms of `tail.rooms`, in that order, with lines missing from the reply: those `tail.gaps` names plus every room +// whose line the byte budget or an unencodable line dropped here. +export function pageOf(persona, deliveries, msgpack, log = () => {}, v3 = false, tail = { lines: [], gaps: [], rooms: [] }) { const page = []; let bytes = 0; for (const delivery of deliveries) { @@ -727,17 +748,20 @@ export function pageOf(persona, deliveries, msgpack, log = () => {}, v3 = false, } if (!v3) return [persona, page]; const lines = []; - for (const line of tail.slice().reverse()) { + const gapped = new Set(tail.gaps); + let full = false; + for (const line of tail.lines.slice().reverse()) { let tuple, size; try { tuple = docs.tailLineTuple(line); size = msgpack.encode(tuple).length; } catch (error) { log(`tail line ${line?.eventId} skipped: ${error?.message ?? error}`); + gapped.add(line?.roomAlias); continue; } - if (bytes + size > docs.limits.replyBytes) break; + if (full || bytes + size > docs.limits.replyBytes) { full = true; gapped.add(line.roomAlias); continue; } lines.unshift(tuple); bytes += size; } - return [persona, page, lines]; + return [persona, page, lines, tail.rooms.filter(room => gapped.has(room))]; } // CultNet RUDP endpoint on loopback, one for every Persona in `mouths` (persona -> mouth). @@ -779,7 +803,7 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey, const request = docs.parsePendingRequest(tuple); const mouth = mouths.get(request.persona); if (!mouth) return refuseUnknown(peer, request.persona); - const tail = v3 ? await mouth.tail().catch(error => { log(`tail not read: ${error?.message ?? error}`); return []; }) : []; + const tail = v3 ? await mouth.tail().catch(error => { log(`tail not read: ${error?.message ?? error}`); const rooms = mouth.speakableRooms(); return { lines: [], gaps: rooms, rooms }; }) : undefined; respond(peer, request.persona, v3 ? docs.kinds.pendingV3 : docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log, v3, tail)); } else if (document.schemaId === docs.kinds.ack) { const request = docs.parseAck(tuple);